import json
import requests
from anthropic import Anthropic
from typing import Dict, Any, List
class AlloyMCPClient:
"""Client for interacting with Alloy MCP Server"""
def __init__(self, server_id: str, access_token: str):
self.base_url = f"https://mcp.runalloy.com/mcp/{server_id}/{access_token}"
def call_tool(self, name: str, arguments: Dict[str, Any] = {}) -> Any:
"""Execute an MCP tool and return the result"""
response = requests.post(
self.base_url,
json={
"jsonrpc": "2.0",
"method": "tools/call",
"params": {"name": name, "arguments": arguments},
"id": 1
},
headers={"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
)
# Parse event-stream response if needed
text = response.text
if text.startswith("data: "):
for line in text.split('\n'):
if line.startswith("data: "):
return json.loads(line[6:]).get("result", {})
return response.json().get("result", {})
# Initialize clients
mcp = AlloyMCPClient(
server_id="your-server-id",
access_token="your-access-token"
)
claude = Anthropic(api_key="your-claude-api-key")
# Define available MCP tools for Claude
tools = [
{
"name": "list_connectors_alloy",
"description": "List all available platform integrations",
"input_schema": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "Optional category filter"
}
}
}
},
{
"name": "execute_action_alloy",
"description": "Execute an action on a connected platform",
"input_schema": {
"type": "object",
"properties": {
"connectorId": {"type": "string"},
"actionId": {"type": "string"},
"parameters": {"type": "object"}
},
"required": ["connectorId", "actionId"]
}
}
]
def chat_with_claude(user_message: str) -> str:
"""Send a message to Claude with MCP tools enabled"""
# Create the conversation
message = claude.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=tools,
messages=[
{"role": "user", "content": user_message}
],
system="You are a helpful assistant that can interact with various platforms through the MCP tools. When users ask about integrations or want to perform actions, use the appropriate tools."
)
# Handle tool use if Claude wants to call a tool
if message.stop_reason == "tool_use":
tool_results = []
for content in message.content:
if content.type == "tool_use":
# Execute the MCP tool
result = mcp.call_tool(content.name, content.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": content.id,
"content": json.dumps(result)
})
# Continue conversation with tool results
followup = claude.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=tools,
messages=[
{"role": "user", "content": user_message},
{"role": "assistant", "content": message.content},
{"role": "user", "content": tool_results}
]
)
return followup.content[0].text
return message.content[0].text
# Example usage
response = chat_with_claude("What Slack actions can I perform?")
print(response)
response = chat_with_claude("Send a message to the #general channel saying 'Hello from Claude!'")
print(response)
