Соедините модели ИИ с живыми данными с помощью Model Context Protocol (MCP)

import json
import asyncio
from dataclasses import dataclass, asdict
from typing import Dict, List, Any, Optional, Callable
from datetime import datetime
import random


@dataclass
class Resource:
   uri: str
   name: str
   description: str
   mime_type: str
   content: Any = None


@dataclass
class Tool:
   name: str
   description: str
   parameters: Dict[str, Any]
   handler: Optional[Callable] = None


@dataclass
class Message:
   role: str
   content: str
   timestamp: str = None
   def __post_init__(self):
       if not self.timestamp:
           self.timestamp = datetime.now().isoformat()

class MCPServer:
   def __init__(self, name: str):
       self.name = name
       self.resources: Dict[str, Resource] = {}
       self.tools: Dict[str, Tool] = {}
       self.capabilities = {"resources": True, "tools": True, "prompts": True, "logging": True}
       print(f"✓ MCP Server '{name}' initialized with capabilities: {list(self.capabilities.keys())}")
   def register_resource(self, resource: Resource) -> None:
       self.resources[resource.uri] = resource
       print(f"  → Resource registered: {resource.name} ({resource.uri})")
   def register_tool(self, tool: Tool) -> None:
       self.tools[tool.name] = tool
       print(f"  → Tool registered: {tool.name}")
   async def get_resource(self, uri: str) -> Optional[Resource]:
       await asyncio.sleep(0.1)
       return self.resources.get(uri)
   async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
       if tool_name not in self.tools:
           raise ValueError(f"Tool '{tool_name}' not found")
       tool = self.tools[tool_name]
       if tool.handler:
           return await tool.handler(**arguments)
       return {"status": "executed", "tool": tool_name, "args": arguments}
   def list_resources(self) -> List[Dict[str, str]]:
       return [{"uri": r.uri, "name": r.name, "description": r.description} for r in self.resources.values()]
   def list_tools(self) -> List[Dict[str, Any]]:
       return [{"name": t.name, "description": t.description, "parameters": t.parameters} for t in self.tools.values()]

class MCPClient:
   def __init__(self, client_id: str):
       self.client_id = client_id
       self.connected_servers: Dict[str, MCPServer] = {}
       self.context: List[Message] = []
       print(f"n✓ MCP Client '{client_id}' initialized")
   def connect_server(self, server: MCPServer) -> None:
       self.connected_servers[server.name] = server
       print(f"  → Connected to server: {server.name}")
   async def query_resources(self, server_name: str) -> List[Dict[str, str]]:
       if server_name not in self.connected_servers:
           raise ValueError(f"Not connected to server: {server_name}")
       return self.connected_servers[server_name].list_resources()
   async def fetch_resource(self, server_name: str, uri: str) -> Optional[Resource]:
       if server_name not in self.connected_servers:
           raise ValueError(f"Not connected to server: {server_name}")
       server = self.connected_servers[server_name]
       resource = await server.get_resource(uri)
       if resource:
           self.add_to_context(Message(role="system", content=f"Fetched resource: {resource.name}"))
       return resource
   async def call_tool(self, server_name: str, tool_name: str, **kwargs) -> Any:
       if server_name not in self.connected_servers:
           raise ValueError(f"Not connected to server: {server_name}")
       server = self.connected_servers[server_name]
       result = await server.execute_tool(tool_name, kwargs)
       self.add_to_context(Message(role="system", content=f"Tool '{tool_name}' executed"))
       return result
   def add_to_context(self, message: Message) -> None:
       self.context.append(message)
   def get_context(self) -> List[Dict[str, Any]]:
       return [asdict(msg) for msg in self.context]

async def analyze_sentiment(text: str) -> Dict[str, Any]:
   await asyncio.sleep(0.2)
   sentiments = ["positive", "negative", "neutral"]
   return {"text": text, "sentiment": random.choice(sentiments), "confidence": round(random.uniform(0.7, 0.99), 2)}


async def summarize_text(text: str, max_length: int = 100) -> Dict[str, str]:
   await asyncio.sleep(0.15)
   summary = text[:max_length] + "..." if len(text) > max_length else text
   return {"original_length": len(text), "summary": summary, "compression_ratio": round(len(summary) / len(text), 2)}


async def search_knowledge(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
   await asyncio.sleep(0.25)
   mock_results = [{"title": f"Result {i+1} for '{query}'", "score": round(random.uniform(0.5, 1.0), 2)} for i in range(top_k)]
   return sorted(mock_results, key=lambda x: x["score"], reverse=True)

async def run_mcp_demo():
   print("=" * 60)
   print("MODEL CONTEXT PROTOCOL (MCP) - ADVANCED TUTORIAL")
   print("=" * 60)
   print("n[1] Setting up MCP Server...")
   server = MCPServer("knowledge-server")
   print("n[2] Registering resources...")
   server.register_resource(Resource(uri="docs://python-guide", name="Python Programming Guide", description="Comprehensive Python documentation", mime_type="text/markdown", content="# Python GuidenPython is a high-level programming language..."))
   server.register_resource(Resource(uri="data://sales-2024", name="2024 Sales Data", description="Annual sales metrics", mime_type="application/json", content={"q1": 125000, "q2": 142000, "q3": 138000, "q4": 165000}))
   print("n[3] Registering tools...")
   server.register_tool(Tool(name="analyze_sentiment", description="Analyze sentiment of text", parameters={"text": {"type": "string", "required": True}}, handler=analyze_sentiment))
   server.register_tool(Tool(name="summarize_text", description="Summarize long text", parameters={"text": {"type": "string", "required": True}, "max_length": {"type": "integer", "default": 100}}, handler=summarize_text))
   server.register_tool(Tool(name="search_knowledge", description="Search knowledge base", parameters={"query": {"type": "string", "required": True}, "top_k": {"type": "integer", "default": 3}}, handler=search_knowledge))
   client = MCPClient("demo-client")
   client.connect_server(server)
   print("n" + "=" * 60)
   print("DEMONSTRATION: MCP IN ACTION")
   print("=" * 60)
   print("n[Demo 1] Listing available resources...")
   resources = await client.query_resources("knowledge-server")
   for res in resources:
       print(f"  • {res['name']}: {res['description']}")
   print("n[Demo 2] Fetching sales data resource...")
   sales_resource = await client.fetch_resource("knowledge-server", "data://sales-2024")
   if sales_resource:
       print(f"  Data: {json.dumps(sales_resource.content, indent=2)}")
   print("n[Demo 3] Analyzing sentiment...")
   sentiment_result = await client.call_tool("knowledge-server", "analyze_sentiment", text="MCP is an amazing protocol for AI integration!")
   print(f"  Result: {json.dumps(sentiment_result, indent=2)}")
   print("n[Demo 4] Summarizing text...")
   summary_result = await client.call_tool("knowledge-server", "summarize_text", text="The Model Context Protocol enables seamless integration between AI models and external data sources...", max_length=50)
   print(f"  Summary: {summary_result['summary']}")
   print("n[Demo 5] Searching knowledge base...")
   search_result = await client.call_tool("knowledge-server", "search_knowledge", query="machine learning", top_k=3)
   print("  Top results:")
   for result in search_result:
       print(f"    - {result['title']} (score: {result['score']})")
   print("n[Demo 6] Current context window...")
   context = client.get_context()
   print(f"  Context length: {len(context)} messages")
   for i, msg in enumerate(context[-3:], 1):
       print(f"  {i}. [{msg['role']}] {msg['content']}")
   print("n" + "=" * 60)
   print("✓ MCP Tutorial Complete!")
   print("=" * 60)
   print("nKey Takeaways:")
   print("• MCP enables modular AI-to-resource connections")
   print("• Resources provide context from external sources")
   print("• Tools enable dynamic operations and actions")
   print("• Async design supports efficient I/O operations")


if __name__ == "__main__":
   import sys
   if 'ipykernel' in sys.modules or 'google.colab' in sys.modules:
       await run_mcp_demo()
   else:
       asyncio.run(run_mcp_demo())