{"skill":{"slug":"mcp-business-integration","displayName":"MCP Business Integration","summary":"Integrate AI agents with business data via Model Context Protocol. Query ads, analytics, CRM data through normalized interfaces. Use when connecting agents t...","description":"---\nname: mcp-integration\ndescription: Integrate AI agents with business data via Model Context Protocol. Query ads, analytics, CRM data through normalized interfaces. Use when connecting agents to business systems, enabling data access, or building MCP servers. Triggers on \"MCP\", \"Model Context Protocol\", \"business data\", \"agent integration\", \"Claude MCP\".\n---\n\n# MCP Integration\n\nModel Context Protocol (MCP) connects AI agents to real business data through normalized interfaces.\n\n## What is MCP?\n\n**Model Context Protocol** is Anthropic's open standard for connecting AI models to external data sources and tools. It provides a unified way for agents to:\n\n- Query databases and APIs\n- Access files and resources\n- Execute tools and functions\n- Maintain context across sessions\n\n## Why MCP Matters\n\n**Before MCP:**\n- Each integration = custom code\n- Different APIs = different patterns\n- Context lost between tools\n- Security = ad-hoc per integration\n\n**With MCP:**\n- One protocol, many integrations\n- Standard patterns for all sources\n- Persistent context\n- Built-in security model\n\n## MCP Architecture\n\n```\n┌─────────────┐     ┌─────────────┐     ┌─────────────┐\n│   Client    │────▶│   Server    │────▶│  Resource   │\n│  (Agent)    │     │   (MCP)     │     │  (Data)     │\n└─────────────┘     └─────────────┘     └─────────────┘\n                           │\n                    ┌──────┴──────┐\n                    │   Tools     │\n                    │  Prompts    │\n                    │  Resources  │\n                    └─────────────┘\n```\n\n### Components\n\n**1. MCP Server**\n- Exposes resources and tools\n- Handles authentication\n- Manages connections\n\n**2. MCP Client**\n- Connects to servers\n- Discovers capabilities\n- Executes operations\n\n**3. Resources**\n- Files, databases, APIs\n- Read/write operations\n- Subscriptions for updates\n\n**4. Tools**\n- Executable functions\n- Input/output schemas\n- Side effects\n\n**5. Prompts**\n- Reusable prompt templates\n- Parameterized\n- Composable\n\n## Integration Types\n\n### 1. Database Integration\n\n```python\n# MCP Server for PostgreSQL\nfrom mcp import Server\n\nserver = Server(\"postgres-integration\")\n\n@server.resource(\"postgres://users\")\nasync def get_users():\n    # Query users from database\n    return await db.query(\"SELECT * FROM users\")\n\n@server.tool(\"query_users\")\nasync def query_users(filters: dict):\n    # Execute parameterized query\n    return await db.query_with_filters(filters)\n```\n\n### 2. API Integration\n\n```python\n# MCP Server for REST API\n@server.resource(\"api://customers\")\nasync def get_customers():\n    response = await httpx.get(\"https://api.example.com/customers\")\n    return response.json()\n\n@server.tool(\"create_customer\")\nasync def create_customer(data: dict):\n    response = await httpx.post(\n        \"https://api.example.com/customers\",\n        json=data\n    )\n    return response.json()\n```\n\n### 3. File System Integration\n\n```python\n# MCP Server for file access\n@server.resource(\"file://documents/{path}\")\nasync def read_document(path: str):\n    with open(f\"documents/{path}\") as f:\n        return f.read()\n\n@server.tool(\"write_document\")\nasync def write_document(path: str, content: str):\n    with open(f\"documents/{path}\", \"w\") as f:\n        f.write(content)\n    return {\"status\": \"written\"}\n```\n\n## Business Data Integration\n\n### Ads Data\n\n```python\n# Google Ads MCP\n@server.resource(\"ads://campaigns\")\nasync def get_campaigns():\n    \"\"\"Get all ad campaigns with metrics\"\"\"\n    campaigns = await ads_client.get_campaigns()\n    return normalize_campaigns(campaigns)\n\n@server.tool(\"optimize_budget\")\nasync def optimize_budget(campaign_id: str):\n    \"\"\"Automatically adjust campaign budget\"\"\"\n    # Analyze performance\n    # Adjust spend allocation\n    # Return optimization results\n```\n\n### Analytics Data\n\n```python\n# Analytics MCP\n@server.resource(\"analytics://metrics\")\nasync def get_metrics():\n    \"\"\"Get normalized metrics across platforms\"\"\"\n    return {\n        \"google_analytics\": await ga.get_metrics(),\n        \"mixpanel\": await mixpanel.get_events(),\n        \"custom_events\": await custom.get_events()\n    }\n\n@server.tool(\"query_analytics\")\nasync def query_analytics(query: str):\n    \"\"\"Natural language analytics query\"\"\"\n    # Parse query\n    # Execute across platforms\n    # Return unified results\n```\n\n### CRM Data\n\n```python\n# Salesforce MCP\n@server.resource(\"crm://leads\")\nasync def get_leads():\n    \"\"\"Get leads from CRM\"\"\"\n    return await salesforce.query(\"SELECT Id, Name, Email FROM Lead\")\n\n@server.tool(\"create_lead\")\nasync def create_lead(data: dict):\n    \"\"\"Create new lead in CRM\"\"\"\n    lead = await salesforce.create(\"Lead\", data)\n    return lead\n```\n\n## Best Practices\n\n### 1. Normalization\n\n```python\n# Normalize data from different sources\ndef normalize_campaign(data, source):\n    schema = {\n        \"id\": data.get(\"id\") or data.get(\"campaign_id\"),\n        \"name\": data.get(\"name\") or data.get(\"campaign_name\"),\n        \"spend\": data.get(\"spend\") or data.get(\"cost\"),\n        \"impressions\": data.get(\"impressions\") or data.get(\"views\"),\n        \"clicks\": data.get(\"clicks\") or data.get(\"clicks_count\"),\n        \"source\": source\n    }\n    return schema\n```\n\n### 2. Error Handling\n\n```python\n@server.tool(\"risky_operation\")\nasync def risky_operation(data: dict):\n    try:\n        result = await external_api.call(data)\n        return {\"success\": True, \"data\": result}\n    except APIError as e:\n        return {\n            \"success\": False,\n            \"error\": str(e),\n            \"suggestion\": \"Try again with valid parameters\"\n        }\n```\n\n### 3. Caching\n\n```python\nfrom functools import lru_cache\nfrom datetime import datetime, timedelta\n\ncache = {}\n\n@server.resource(\"api://expensive-data\")\nasync def get_expensive_data():\n    cache_key = \"expensive-data\"\n    cached = cache.get(cache_key)\n    \n    if cached and cached[\"expires\"] > datetime.now():\n        return cached[\"data\"]\n    \n    # Fetch fresh data\n    data = await expensive_api_call()\n    cache[cache_key] = {\n        \"data\": data,\n        \"expires\": datetime.now() + timedelta(hours=1)\n    }\n    return data\n```\n\n### 4. Security\n\n```python\n# Validate inputs\nfrom pydantic import BaseModel\n\nclass QueryInput(BaseModel):\n    table: str\n    filters: dict\n    limit: int = 100\n\n@server.tool(\"safe_query\")\nasync def safe_query(input: QueryInput):\n    # Input is validated by Pydantic\n    # SQL injection prevented\n    return await db.query(input.table, input.filters, input.limit)\n```\n\n## Claude Desktop Integration\n\n```json\n// claude_desktop_config.json\n{\n  \"mcpServers\": {\n    \"business-data\": {\n      \"command\": \"python\",\n      \"args\": [\"mcp_server.py\"],\n      \"env\": {\n        \"DATABASE_URL\": \"postgresql://...\",\n        \"API_KEY\": \"...\"\n      }\n    }\n  }\n}\n```\n\n## Common MCP Servers\n\n### Official Servers\n\n| Server | Description |\n|--------|-------------|\n| filesystem | File system access |\n| postgres | PostgreSQL database |\n| sqlite | SQLite database |\n| github | GitHub API |\n| google-drive | Google Drive |\n| slack | Slack API |\n\n### Custom Servers\n\nCreate custom servers for:\n- Internal APIs\n- Proprietary databases\n- Custom tools\n- Business-specific operations\n\n## Debugging\n\n### Server Logs\n\n```python\nimport logging\n\nlogging.basicConfig(level=logging.DEBUG)\nlogger = logging.getLogger(\"mcp_server\")\n\n@server.tool(\"debug_operation\")\nasync def debug_operation(data: dict):\n    logger.debug(f\"Input: {data}\")\n    result = await process(data)\n    logger.debug(f\"Output: {result}\")\n    return result\n```\n\n### Connection Issues\n\n```bash\n# Test MCP server\npython -m mcp.server --debug\n\n# Test client connection\npython -m mcp.client --url \"ws://localhost:8080\"\n```\n\n## Examples\n\n### Query Multiple Data Sources\n\n```python\n@server.tool(\"cross_platform_query\")\nasync def cross_platform_query(query: str):\n    \"\"\"Query across multiple platforms\"\"\"\n    results = {}\n    \n    # Query each platform\n    results[\"analytics\"] = await analytics.query(query)\n    results[\"crm\"] = await crm.query(query)\n    results[\"ads\"] = await ads.query(query)\n    \n    # Merge results\n    return merge_results(results)\n```\n\n### Automated Insights\n\n```python\n@server.tool(\"generate_insights\")\nasync def generate_insights(data_source: str):\n    \"\"\"Generate insights from business data\"\"\"\n    # Get data\n    data = await get_data(data_source)\n    \n    # Analyze\n    insights = []\n    \n    # Trend analysis\n    if data[\"trend\"] == \"increasing\":\n        insights.append(\"Revenue trending up - consider scaling\")\n    \n    # Anomaly detection\n    if data[\"anomaly\"]:\n        insights.append(f\"Anomaly detected: {data['anomaly']}\")\n    \n    return {\"insights\": insights, \"data\": data}\n```\n\n## Resources\n\n- **Anthropic MCP Docs:** https://modelcontextprotocol.io\n- **Official Servers:** https://github.com/modelcontextprotocol/servers\n- **Community Servers:** https://github.com/punkpeye/awesome-mcp-servers\n","tags":{"analytics":"1.0.0","business-data":"1.0.0","crm":"1.0.0","latest":"1.0.0","mcp":"1.0.0","protocol":"1.0.0"},"stats":{"comments":0,"downloads":558,"installsAllTime":0,"installsCurrent":0,"stars":0,"versions":1},"createdAt":1773682626298,"updatedAt":1778491953611},"latestVersion":{"version":"1.0.0","createdAt":1773682626298,"changelog":"Initial release","license":"MIT-0"},"metadata":null,"owner":{"handle":"engsathiago","userId":"s1748ca82aw6dtzhdqmtcdaj1h83fxeq","displayName":"engsathiago","image":"https://avatars.githubusercontent.com/u/40508117?v=4"},"moderation":null}