Claude API i MCP - budowanie agentów AI z narzędziami

pip install anthropic>=0.18.0 mcp-client>=2.1.0
pip install pydantic>=2.0.0 asyncio-mqtt sqlite3

import anthropic
import asyncio
from mcp_client import MCPClient
from pydantic import BaseModel
from typing import Dict, List, Optional

client = anthropic.Anthropic(
api_key=”your-api-key-here”
)

# Definicja narzędzia do sprawdzania pogody
weather_tool = {
„name”: „get_weather”,
„description”: „Pobiera aktualną pogodę dla podanego miasta”,
„input_schema”: {
„type”: „object”,
„properties”: {
„city”: {„type”: „string”, „description”: „Nazwa miasta”},
„units”: {„type”: „string”, „enum”: [„celsius”, „fahrenheit”]}
},
„required”: [„city”]
}
}

def get_weather(city: str, units: str = „celsius”) -> str:
# Symulacja API pogodowego
weather_data = {
„Warszawa”: {„temp”: 15, „condition”: „słonecznie”},
„Kraków”: {„temp”: 12, „condition”: „pochmurnie”}
}

if city in weather_data:
data = weather_data[city]
return f”W {city}: {data[’temp’]}°{’C’ if units == 'celsius’ else 'F’}, {data[’condition’]}”
else:
return f”Brak danych pogodowych dla miasta {city}”

# Wywołanie z function calling
response = client.messages.create(
model=”claude-3-opus-20240229″,
max_tokens=1024,
tools=[weather_tool],
messages=[{„role”: „user”, „content”: „Jaka jest pogoda w Warszawie?”}]
)

from pydantic import BaseModel
from typing import List

class TaskAnalysis(BaseModel):
priority: str # „high”, „medium”, „low”
category: str
estimated_hours: float
required_skills: List[str]
dependencies: List[str]

system_prompt = „””
Analizujesz zadania projektowe i zwracasz ustrukturalizowane dane.
Odpowiadaj WYŁĄCZNIE w formacie JSON zgodnym ze schematem TaskAnalysis.
Nie dodawaj komentarzy ani dodatkowego tekstu.
„””

user_message = „Stwórz aplikację mobilną do zarządzania budżetem domowym z synchronizacją w chmurze”

response = client.messages.create(
model=”claude-3-sonnet-20240229″,
max_tokens=512,
system=system_prompt,
messages=[{„role”: „user”, „content”: user_message}]
)

import json
from mcp_client import MCPClient

# Konfiguracja MCP Server
mcp_config = {
„servers”: {
„database_server”: {
„command”: „python”,
„args”: [„mcp_servers/database_server.py”],
„env”: {„DATABASE_URL”: „sqlite:///app.db”}
},
„file_server”: {
„command”: „python”,
„args”: [„mcp_servers/file_server.py”],
„env”: {„WORKSPACE_PATH”: „/workspace”}
}
}
}

# Inicjalizacja klienta MCP
async def init_mcp_client():
client = MCPClient()
await client.connect(„database_server”)
await client.connect(„file_server”)
return client

class SalesAnalysisAgent:
def __init__(self, anthropic_client, mcp_client):
self.claude = anthropic_client
self.mcp = mcp_client
self.tools = [
{
„name”: „query_sales_data”,
„description”: „Wykonuje zapytania SQL do bazy danych sprzedażowych”,
„input_schema”: {
„type”: „object”,
„properties”: {
„query”: {„type”: „string”, „description”: „Zapytanie SQL”},
„date_range”: {„type”: „string”, „description”: „Zakres dat (YYYY-MM-DD to YYYY-MM-DD)”}
},
„required”: [„query”]
}
},
{
„name”: „generate_chart”,
„description”: „Tworzy wykresy na podstawie danych”,
„input_schema”: {
„type”: „object”,
„properties”: {
„chart_type”: {„type”: „string”, „enum”: [„bar”, „line”, „pie”]},
„data”: {„type”: „array”, „description”: „Dane do wizualizacji”},
„title”: {„type”: „string”, „description”: „Tytuł wykresu”}
},
„required”: [„chart_type”, „data”, „title”]
}
}
]

async def analyze_sales(self, question: str):
system_prompt = „””
Jesteś ekspertem od analizy danych sprzedażowych.
Masz dostęp do bazy danych i narzędzi do wizualizacji.

Gdy otrzymasz pytanie:
1. Zastanów się jakie dane są potrzebne
2. Wykonaj odpowiednie zapytania SQL
3. Przeanalizuj wyniki
4. Wygeneruj wizualizacje jeśli to pomoże
5. Przedstaw wnioski w czytelny sposób

Zawsze uzasadniaj swoje analizy konkretnymi danymi.
„””

response = await self.claude.messages.create(
model=”claude-3-opus-20240229″,
max_tokens=2048,
system=system_prompt,
tools=self.tools,
messages=[{„role”: „user”, „content”: question}]
)

return response

import logging
from typing import Optional

class ErrorHandlingAgent:
def __init__(self, claude_client, max_retries=3):
self.claude = claude_client
self.max_retries = max_retries
self.logger = logging.getLogger(__name__)

async def safe_function_call(self, function_name: str, parameters: dict) -> Optional[dict]:
„””Bezpieczne wywołanie funkcji z obsługą błędów”””
for attempt in range(self.max_retries):
try:
# Walidacja parametrów przed wywołaniem
validated_params = self.validate_parameters(function_name, parameters)
result = await self.execute_function(function_name, validated_params)
return result

except ValidationError as e:
self.logger.error(f”Błąd walidacji parametrów: {e}”)
return {„error”: f”Nieprawidłowe parametry: {str(e)}”}

except TimeoutError:
self.logger.warning(f”Timeout przy próbie {attempt + 1}”)
if attempt == self.max_retries - 1:
return {„error”: „Przekroczono limit czasu wykonania”}
await asyncio.sleep(2 ** attempt) # Exponential backoff

except Exception as e:
self.logger.error(f”Nieoczekiwany błąd: {e}”)
return {„error”: f”Błąd wykonania: {str(e)}”}

chain_of_thought_prompt = „””
Jesteś agentem do planowania projektów. Gdy otrzymasz opis zadania:

1. Przeanalizuj wymagania i zidentyfikuj główne komponenty
2. Określ zależności między zadaniami
3. Oszacuj czas wykonania poszczególnych etapów
4. Zidentyfikuj potencjalne ryzyka i wąskie gardła
5. Zaproponuj harmonogram z buforami czasowymi

Następnie przedstaw szczegółowy plan z uzasadnieniem każdego kroku.
Korzystaj z dostępnych narzędzi do zarządzania projektami.

Pamiętaj: lepiej przewidzieć więcej czasu niż nie dotrzymać terminów.
„””

user_query = „Zaplanuj migrację 500 użytkowników z systemu legacy do nowej platformy CRM”

response = await client.messages.create(
model=”claude-3-opus-20240229″,
max_tokens=3000,
system=chain_of_thought_prompt,
tools=project_management_tools,
messages=[{„role”: „user”, „content”: user_query}]
)

# Implementacja multi-step reasoning
class ReasoningAgent:
async def solve_complex_problem(self, problem: str):
steps = [
„analysis”, # Analiza problemu
„planning”, # Planowanie rozwiązania
„execution”, # Wykonanie kroków
„validation” # Walidacja wyników
]

context = {„problem”: problem, „intermediate_results”: []}

for step in steps:
step_prompt = self.generate_step_prompt(step, context)

response = await self.claude.messages.create(
model=”claude-3-sonnet-20240229″,
max_tokens=1024,
system=step_prompt,
tools=self.get_tools_for_step(step),
messages=self.build_context_messages(context)
)

context[„intermediate_results”].append({
„step”: step,
„result”: response.content[0].text,
„tool_calls”: [call for call in response.content if call.type == „tool_use”]
})

return context

import time
import asyncio
from dataclasses import dataclass
from typing import List

@dataclass
class PerformanceMetrics:
request_count: int
total_tokens: int
total_cost: float
average_response_time: float
error_rate: float

class MonitoringAgent:
def __init__(self, claude_client):
self.claude = claude_client
self.metrics = PerformanceMetrics(0, 0, 0.0, 0.0, 0.0)
self.response_times = []
self.error_count = 0

async def tracked_request(self, model: str, messages: list, tools: list = None):
start_time = time.time()

try:
response = await self.claude.messages.create(
model=model,
max_tokens=1024,
messages=messages,
tools=tools or []
)

# Aktualizacja metryk
response_time = time.time() - start_time
self.response_times.append(response_time)

self.metrics.request_count += 1
self.metrics.total_tokens += response.usage.input_tokens + response.usage.output_tokens
self.metrics.total_cost += self.calculate_cost(response.usage, model)
self.metrics.average_response_time = sum(self.response_times) / len(self.response_times)

return response

except Exception as e:
self.error_count += 1
self.metrics.error_rate = self.error_count / (self.metrics.request_count + self.error_count)
raise e