442 lines
No EOL
11 KiB
Text
442 lines
No EOL
11 KiB
Text
---
|
|
title: "Integrate MCP into Your Python App"
|
|
description: "Add MCP support to your LLM application in Python"
|
|
---
|
|
|
|
Learn how to integrate MCP into your Python LLM application. We'll build a simple web analyzer that uses Claude to analyze websites, enhanced with MCP's web automation capabilities.
|
|
|
|
<Note>
|
|
This guide uses an existing MCP server ([puppeteer](/servers/puppeteer)) rather than building one from scratch. This lets us focus on the integration patterns.
|
|
</Note>
|
|
|
|
## The example app
|
|
|
|
We'll build an "AI Web Analyzer" that:
|
|
1. Takes a URL from the user
|
|
2. Captures the webpage using MCP
|
|
3. Has the LLM analyze the page's content, design, and accessibility
|
|
|
|
<Frame>
|
|
<img src="/api/placeholder/800/400" alt="Screenshot of the Web Analyzer CLI interface" />
|
|
</Frame>
|
|
|
|
## Prerequisites
|
|
|
|
<Steps>
|
|
<Step title="Install dependencies">
|
|
```bash
|
|
pip install modelcontextprotocol-sdk anthropic click rich
|
|
```
|
|
</Step>
|
|
|
|
<Step title="Set up puppeteer server">
|
|
```bash
|
|
# Install example MCP servers
|
|
git clone https://github.com/modelcontextprotocol/servers
|
|
cd servers
|
|
npm install
|
|
npm run build
|
|
npm install -g
|
|
```
|
|
</Step>
|
|
|
|
<Step title="Get API key">
|
|
```bash
|
|
# Set your Anthropic API key
|
|
export ANTHROPIC_API_KEY='your-key-here'
|
|
```
|
|
</Step>
|
|
</Steps>
|
|
|
|
## Building the base app
|
|
|
|
First, let's create a basic version without MCP:
|
|
|
|
<Steps>
|
|
<Step title="Create project structure">
|
|
```bash
|
|
mkdir web-analyzer
|
|
cd web-analyzer
|
|
touch analyzer.py
|
|
```
|
|
</Step>
|
|
|
|
<Step title="Add base implementation">
|
|
```python
|
|
# analyzer.py
|
|
import click
|
|
import anthropic
|
|
from rich.console import Console
|
|
from rich.panel import Panel
|
|
from rich import print as rprint
|
|
|
|
console = Console()
|
|
client = anthropic.Client()
|
|
|
|
def analyze_url(url: str) -> str:
|
|
messages = [{
|
|
"role": "user",
|
|
"content": f"Please analyze this website: {url}\n\n"
|
|
f"I'd like to know about:\n"
|
|
f"1. Content quality and organization\n"
|
|
f"2. Visual design (based on the URL/description)\n"
|
|
f"3. Potential improvements"
|
|
}]
|
|
|
|
response = client.messages.create(
|
|
model="claude-3-opus-20240229",
|
|
max_tokens=1000,
|
|
messages=messages
|
|
)
|
|
return response.content[0].text
|
|
|
|
@click.command()
|
|
@click.argument('url')
|
|
def main(url: str):
|
|
with console.status("[bold green]Analyzing website..."):
|
|
analysis = analyze_url(url)
|
|
|
|
rprint(Panel.fit(
|
|
analysis,
|
|
title="Website Analysis",
|
|
border_style="blue"
|
|
))
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
```
|
|
</Step>
|
|
|
|
<Step title="Test the base app">
|
|
```bash
|
|
python analyzer.py https://example.com
|
|
```
|
|
</Step>
|
|
</Steps>
|
|
|
|
## Adding MCP support
|
|
|
|
Now let's enhance the app with MCP capabilities:
|
|
|
|
<Steps>
|
|
<Step title="Create MCP client">
|
|
```python
|
|
# At the top of analyzer.py
|
|
from modelcontextprotocol.client import Client
|
|
from modelcontextprotocol.client.stdio import StdioClientTransport
|
|
import asyncio
|
|
import json
|
|
from typing import Optional
|
|
|
|
class MCPWebClient:
|
|
def __init__(self):
|
|
self.client = Client(
|
|
client_info={
|
|
"name": "web-analyzer",
|
|
"version": "0.1.0"
|
|
}
|
|
)
|
|
self.transport = None
|
|
|
|
async def connect(self):
|
|
self.transport = StdioClientTransport(
|
|
server={
|
|
"command": "mcp-server-puppeteer",
|
|
}
|
|
)
|
|
await self.client.connect(self.transport)
|
|
|
|
async def navigate(self, url: str) -> Optional[str]:
|
|
try:
|
|
result = await self.client.request(
|
|
{
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": "navigate",
|
|
"arguments": {"url": url}
|
|
}
|
|
},
|
|
schema=CallToolResultSchema
|
|
)
|
|
return result.get("toolResult")
|
|
except Exception as e:
|
|
console.print(f"[red]Error navigating to URL: {e}")
|
|
return None
|
|
|
|
async def get_console_logs(self) -> str:
|
|
result = await self.client.request(
|
|
{
|
|
"method": "resources/read",
|
|
"params": {"uri": "console://logs"}
|
|
},
|
|
schema=ReadResourceResultSchema
|
|
)
|
|
return result["contents"][0]["text"]
|
|
|
|
async def close(self):
|
|
if self.transport:
|
|
await self.client.close()
|
|
```
|
|
</Step>
|
|
<Step title="Update the analyzer">
|
|
```python
|
|
async def analyze_url_with_mcp(url: str) -> str:
|
|
# Create and connect MCP client
|
|
mcp = MCPWebClient()
|
|
await mcp.connect()
|
|
# Navigate to the URL
|
|
await mcp.navigate(url)
|
|
|
|
# Get console logs (includes errors, warnings, etc.)
|
|
console_logs = await mcp.get_console_logs()
|
|
|
|
# Close the connection
|
|
await mcp.close()
|
|
|
|
# Analyze with Claude
|
|
messages = [{
|
|
"role": "user",
|
|
"content": (
|
|
f"Please analyze this website: {url}\n\n"
|
|
f"Here are the browser console logs:\n{console_logs}\n\n"
|
|
f"Please analyze:\n"
|
|
f"1. Content quality and organization\n"
|
|
f"2. Visual design\n"
|
|
f"3. Technical issues (based on console logs)\n"
|
|
f"4. Potential improvements"
|
|
)
|
|
}]
|
|
|
|
response = client.messages.create(
|
|
model="claude-3-opus-20240229",
|
|
max_tokens=1000,
|
|
messages=messages
|
|
)
|
|
return response.content[0].text
|
|
|
|
@click.command()
|
|
@click.argument('url')
|
|
def main(url: str):
|
|
with console.status("[bold green]Analyzing website..."):
|
|
analysis = asyncio.run(analyze_url_with_mcp(url))
|
|
|
|
rprint(Panel.fit(
|
|
analysis,
|
|
title="Website Analysis",
|
|
border_style="blue"
|
|
))
|
|
```
|
|
</Step>
|
|
<Step title="Run the enhanced version">
|
|
```bash
|
|
python analyzer.py https://example.com
|
|
```
|
|
</Step>
|
|
</Steps>
|
|
|
|
## Understanding MCP integration
|
|
|
|
Let's break down the key components of MCP integration:
|
|
<Tabs>
|
|
<Tab title="Client Setup">
|
|
```python
|
|
class MCPWebClient:
|
|
def __init__(self):
|
|
self.client = Client(
|
|
client_info={
|
|
"name": "web-analyzer",
|
|
"version": "0.1.0"
|
|
}
|
|
)
|
|
```
|
|
The MCP client identifies your application and manages the connection to MCP servers.
|
|
</Tab>
|
|
<Tab title="Transport">
|
|
```python
|
|
self.transport = StdioClientTransport(
|
|
server={
|
|
"command": "mcp-server-puppeteer",
|
|
}
|
|
)
|
|
```
|
|
The transport layer handles communication with the MCP server. Here we're using stdio to talk to the puppeteer server.
|
|
</Tab>
|
|
<Tab title="Tools">
|
|
```python
|
|
result = await self.client.request(
|
|
{
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": "navigate",
|
|
"arguments": {"url": url}
|
|
}
|
|
},
|
|
schema=CallToolResultSchema
|
|
)
|
|
```
|
|
Tools let you take actions through MCP servers. Here we're using puppeteer's navigate tool.
|
|
</Tab>
|
|
<Tab title="Resources">
|
|
```python
|
|
result = await self.client.request(
|
|
{
|
|
"method": "resources/read",
|
|
"params": {"uri": "console://logs"}
|
|
},
|
|
schema=ReadResourceResultSchema
|
|
)
|
|
```
|
|
Resources provide data from MCP servers. Here we're reading console logs that puppeteer collected.
|
|
</Tab>
|
|
</Tabs>
|
|
|
|
## Enhancement ideas
|
|
|
|
Now that you have MCP working, here are some ways to enhance the analyzer:
|
|
<CardGroup cols={2}>
|
|
<Card
|
|
title="Visual Analysis"
|
|
icon="image"
|
|
>
|
|
```python
|
|
# Take screenshots with puppeteer
|
|
await client.request({
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": "screenshot",
|
|
"arguments": {...}
|
|
}
|
|
})
|
|
```
|
|
</Card>
|
|
<Card
|
|
title="Multiple Pages"
|
|
icon="layer-group">
|
|
```python
|
|
# Analyze multiple pages in sequence
|
|
for url in urls:
|
|
await navigate(url)
|
|
await analyze()
|
|
```
|
|
</Card>
|
|
<Card
|
|
title="Interactive Elements"
|
|
icon="mouse-pointer">
|
|
```python
|
|
# Test interactive features
|
|
await client.request({
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": "click",
|
|
"arguments": {"selector": "button"}
|
|
}
|
|
})
|
|
```
|
|
</Card>
|
|
</CardGroup>
|
|
|
|
Common Patterns
|
|
Here are some patterns you'll often use when integrating MCP:
|
|
<AccordionGroup>
|
|
<Accordion title="Resource Polling" active>
|
|
When you need to wait for resource updates:
|
|
```python
|
|
async def wait_for_resource(uri: str, timeout: int = 30):
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
result = await client.request(
|
|
{
|
|
"method": "resources/read",
|
|
"params": {"uri": uri}
|
|
},
|
|
schema=ReadResourceResultSchema
|
|
)
|
|
if some_condition(result):
|
|
return result
|
|
await asyncio.sleep(1)
|
|
raise TimeoutError()
|
|
```
|
|
</Accordion>
|
|
<Accordion title="Error Handling">
|
|
Robust error handling for MCP operations:
|
|
```python
|
|
try:
|
|
await client.request(...)
|
|
except MCPError as e:
|
|
if e.code == ErrorCode.ResourceNotFound:
|
|
# Handle missing resource
|
|
elif e.code == ErrorCode.InvalidRequest:
|
|
# Handle invalid parameters
|
|
else:
|
|
# Handle other errors
|
|
```
|
|
</Accordion>
|
|
<Accordion title="Resource Subscriptions">
|
|
Subscribe to resource updates:
|
|
```python
|
|
await client.request(
|
|
{
|
|
"method": "resources/subscribe",
|
|
"params": {"uri": "console://logs"}
|
|
},
|
|
schema=EmptyResultSchema
|
|
)
|
|
# Now you'll get notifications when logs change
|
|
```
|
|
</Accordion>
|
|
</AccordionGroup>
|
|
|
|
## Troubleshooting
|
|
|
|
<AccordionGroup>
|
|
<Accordion title="Connection Issues">
|
|
```python
|
|
# Enable debug logging
|
|
import logging
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
```
|
|
Watch for connection errors and MCP protocol messages.
|
|
</Accordion>
|
|
<Accordion title="Resource Errors">
|
|
1. Check if the resource exists:
|
|
```python
|
|
resources = await client.request(
|
|
{"method": "resources/list"},
|
|
schema=ListResourcesResultSchema
|
|
)
|
|
print(resources)
|
|
```
|
|
2. Verify URI format
|
|
3. Check server logs
|
|
</Accordion>
|
|
<Accordion title="Tool Errors">
|
|
1. List available tools:
|
|
```python
|
|
tools = await client.request(
|
|
{"method": "tools/list"},
|
|
schema=ListToolsResultSchema
|
|
)
|
|
print(tools)
|
|
```
|
|
2. Check argument schema
|
|
3. Verify server capabilities
|
|
</Accordion>
|
|
</AccordionGroup>
|
|
|
|
## Next steps
|
|
|
|
<CardGroup cols={2}>
|
|
<Card
|
|
title="Add More Servers"
|
|
icon="plus"
|
|
href="/servers"
|
|
>
|
|
Connect to additional MCP servers for more capabilities
|
|
</Card>
|
|
<Card
|
|
title="Build Custom Server"
|
|
icon="code"
|
|
href="/docs/guides/first-server-python">
|
|
Create your own MCP server for custom functionality
|
|
</Card>
|
|
</CardGroup> |