Tool and API integration via Model Context Protocol

import os
import json
from typing import List, Dict, Any

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic

os.environ["ANTHROPIC_API_KEY"] = "your_anthropic_api_key_here"

client = Anthropic()

# Absolute path to your running MCP server script
mcp_server_path = "absolute/path/to/your/running/mcp/server"

async def discover_tools():
    """Connect to the MCP server and discover available tools."""
    server_params = StdioServerParameters(
        command="python",
        args=[mcp_server_path],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()

            tool_info = []
            for tool_type, tool_list in tools:
                if tool_type == "tools":
                    for tool in tool_list:
                        tool_info.append({
                            "name": tool.name,
                            "description": tool.description,
                            "schema": tool.inputSchema
                        })

            return tool_info

async def execute_tool(tool_name: str, arguments: Dict[str, Any]):
    """Execute a specific tool provided by the MCP server."""
    server_params = StdioServerParameters(
        command="python",
        args=[mcp_server_path],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            result = await session.call_tool(tool_name, arguments)
            return result

async def query_claude(
    prompt: str,
    tool_info: List[Dict],
    previous_messages=None
):
    """Send a query to Claude and process the response, executing tools as needed."""
    if previous_messages is None:
        previous_messages = []

    # Format tool information for Claude's system prompt
    tool_descriptions = "\n\n".join([
        f"Tool: {tool['name']}\n"
        f"Description: {tool['description']}\n"
        f"Schema: {json.dumps(tool['schema'], indent=2)}"
        for tool in tool_info
    ])

    system_prompt = f"""You are an AI assistant with access to specialized tools through MCP.

Available tools:
{tool_descriptions}

When you need to use a tool, respond with a JSON object in the following format:
{{
    "tool": "tool_name",
    "arguments": {{
        "arg1": "value1"
    }}
}}

Do not include any other text when using a tool, just the JSON object.
For regular responses, simply respond normally.
"""

    messages = [msg for msg in previous_messages if msg["role"] != "system"]
    messages.append({"role": "user", "content": prompt})

    response = client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=4000,
        system=system_prompt,
        messages=messages
    )

    claude_response = response.content[0].text

    # Detect and execute tool calls embedded in Claude's response
    try:
        import re
        json_match = re.search(r'(\{[\s\S]*\})', claude_response)

        if json_match:
            tool_request = json.loads(json_match.group(1))

            if "tool" in tool_request and "arguments" in tool_request:
                tool_result = await execute_tool(
                    tool_request["tool"],
                    tool_request["arguments"]
                )

                if not isinstance(tool_result, str):
                    tool_result = str(tool_result)

                messages.append({"role": "assistant", "content": claude_response})
                messages.append({"role": "user", "content": f"Tool result: {tool_result}"})

                final_response = client.messages.create(
                    model="claude-3-5-sonnet-20240620",
                    max_tokens=4000,
                    system=system_prompt,
                    messages=messages
                )

                return final_response.content[0].text, messages

    except (json.JSONDecodeError, KeyError, AttributeError):
        pass

    return claude_response, messages

# First, discover the tools
tools = await discover_tools()
print(f"Discovered {len(tools)} tools:")
for tool in tools:
    print(f"  {tool['name']}: {tool['description']}")

# Then query the agent
response, messages = await query_claude(
    "What is the current price of Bitcoin?",
    tools
)
print("Response:", response)

async def chat_session():
    """Run an interactive chat session with the MCP agent."""
    tools = await discover_tools()
    print(f"Agent ready with {len(tools)} tools.")

    messages = []

    while True:
        user_input = input("\nYou: ")
        if user_input.lower() in ["exit", "quit"]:
            break

        response, messages = await query_claude(user_input, tools, messages)
        print(f"Assistant: {response}")

# Start the session
await chat_session()