pip install mcp

# mcp_server.py — expose your tools to any MCP client
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio as stdio
from mcp.server.models import InitializationOptions
import asyncio

server = Server("my-enterprise-tools")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="search_knowledge_base",
            description="Search internal documentation. USE for product/process questions. DO NOT USE for general knowledge.",
            inputSchema={
                "type": "object",
                "properties": {"query": {"type": "string", "description": "Natural language search query"}},
                "required": ["query"]
            }
        ),
        Tool(
            name="create_jira_ticket",
            description="Create a Jira bug ticket. USE when user reports a software defect.",
            inputSchema={
                "type": "object",
                "properties": {
                    "summary": {"type": "string"},
                    "description": {"type": "string"},
                    "priority": {"type": "string", "enum": ["Low","Medium","High","Critical"]}
                },
                "required": ["summary", "description"]
            }
        ),
        Tool(
            name="get_user_info",
            description="Look up employee information by name or email.",
            inputSchema={
                "type": "object",
                "properties": {"identifier": {"type": "string"}},
                "required": ["identifier"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "search_knowledge_base":
        # Call your RAG pipeline
        results = await rag_search(arguments["query"])
        return [TextContent(type="text", text=str(results))]
    elif name == "create_jira_ticket":
        ticket = await create_jira(arguments)
        return [TextContent(type="text", text=f"Created: {ticket['key']}")]
    elif name == "get_user_info":
        user = await lookup_user(arguments["identifier"])
        return [TextContent(type="text", text=str(user))]

async def main():
    async with stdio.stdio_server() as (read, write):
        await server.run(read, write, InitializationOptions(server_name="my-enterprise-tools"))

if __name__ == "__main__":
    asyncio.run(main())

# Register in ~/.config/claude/claude_desktop_config.json:
# {
#   "mcpServers": {
#     "my-enterprise-tools": {
#       "command": "python",
#       "args": ["/path/to/mcp_server.py"]
#     }
#   }
# }

# Run n8n locally
docker run -it --rm -p 5678:5678 -v ~/.n8n:/home/node/.n8n n8nio/n8n
# Open http://localhost:5678

# n8n workflow: AI email triage
# [Gmail trigger: new email arrives]
#   -> [HTTP Request: POST to your FastAPI /classify {subject, body}]
#   -> [Switch node: route on classification]
#       -> billing: [Stripe API: fetch customer] -> [Gmail: send billing reply]
#       -> technical: [Jira: create ticket] -> [Slack: notify #eng-support]
#       -> general: [HTTP: POST /generate-reply] -> [Gmail: send AI reply]

# Your FastAPI endpoint that n8n calls:
from fastapi import FastAPI
from pydantic import BaseModel
import instructor, anthropic
from typing import Literal

app = FastAPI()
instr_client = instructor.from_anthropic(anthropic.Anthropic())

class EmailRequest(BaseModel):
    subject: str
    body: str

class ClassifyResponse(BaseModel):
    category: Literal["billing", "technical", "general", "complaint"]
    confidence: float
    suggested_response: str

@app.post("/classify")
async def classify_email(request: EmailRequest) -> ClassifyResponse:
    return instr_client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=256,
        messages=[{"role": "user", "content":
            f"Classify this support email and suggest a brief response.\n\n"
            f"Subject: {request.subject}\nBody: {request.body[:500]}"}],
        response_model=ClassifyResponse
    )

@app.post("/generate-reply")
async def generate_reply(request: EmailRequest) -> dict:
    # Use RAG to ground the reply in documentation
    context = await rag_search(request.subject + " " + request.body[:200])
    response = anthropic.Anthropic().messages.create(
        model="claude-3-5-sonnet-20241022", max_tokens=512,
        messages=[{"role": "user", "content":
            f"Write a professional reply to this email.\n\n"
            f"Context: {context}\n\nEmail: Subject: {request.subject}\n{request.body}"}]
    )
    return {"reply": response.content[0].text}

pip install slack-sdk slack-bolt

from slack_bolt import App
from slack_bolt.adapter.fastapi import SlackRequestHandler
import anthropic

slack_app = App(token=os.environ["SLACK_BOT_TOKEN"],
                signing_secret=os.environ["SLACK_SIGNING_SECRET"])
handler = SlackRequestHandler(slack_app)
client = anthropic.Anthropic()

@slack_app.event("app_mention")
def handle_mention(event, say, client_slack):
    user_message = event["text"]
    channel = event["channel"]
    user = event["user"]

    # Show typing indicator
    client_slack.reactions_add(channel=channel, timestamp=event["ts"], name="thinking_face")

    # Classify and respond
    category = classify_request(user_message)

    if category == "technical":
        # RAG-powered answer
        result = rag_query(user_message)
        response = result["answer"]
        sources = "\n".join([f"• {s['source']}" for s in result["sources"][:3]])
        say(f"{response}\n\nSources:\n{sources}", thread_ts=event["ts"])

    elif category == "ticket":
        # Create Jira ticket
        ticket = create_jira_ticket(
            summary=user_message[:80],
            description=f"Reported by <@{user}> in Slack:\n{user_message}"
        )
        say(f"Created Jira ticket: *{ticket['key']}*\n{ticket['url']}",
            thread_ts=event["ts"])

    else:
        # General AI response
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022", max_tokens=512,
            messages=[{"role": "user", "content": user_message}]
        )
        say(response.content[0].text, thread_ts=event["ts"])

    # Remove thinking indicator
    client_slack.reactions_remove(channel=channel, timestamp=event["ts"], name="thinking_face")

# FastAPI handler for Slack events
from fastapi import FastAPI, Request
api = FastAPI()

@api.post("/slack/events")
async def endpoint(req: Request):
    return await handler.handle(req)

import requests
import os

# ── Jira: create tickets from AI classification ────────
def create_jira_ticket(summary: str, description: str,
                        priority: str = "Medium", project: str = "ENG") -> dict:
    response = requests.post(
        f"{os.environ['JIRA_BASE_URL']}/rest/api/3/issue",
        auth=(os.environ['JIRA_EMAIL'], os.environ['JIRA_API_TOKEN']),
        json={
            "fields": {
                "project": {"key": project},
                "summary": summary,
                "description": {
                    "type": "doc", "version": 1,
                    "content": [{"type": "paragraph",
                                 "content": [{"type": "text", "text": description}]}]
                },
                "issuetype": {"name": "Bug"},
                "priority": {"name": priority}
            }
        }
    )
    data = response.json()
    return {"key": data["key"], "url": f"{os.environ['JIRA_BASE_URL']}/browse/{data['key']}"}

# ── Notion: append AI summary to a page ───────────────
def append_to_notion_page(page_id: str, ai_summary: str):
    requests.patch(
        f"https://api.notion.com/v1/blocks/{page_id}/children",
        headers={"Authorization": f"Bearer {os.environ['NOTION_TOKEN']}",
                 "Notion-Version": "2022-06-28"},
        json={"children": [{"type": "callout", "callout": {
            "rich_text": [{"type": "text", "text": {"content": f"AI Summary: {ai_summary}"}}],
            "icon": {"emoji": "🤖"}
        }}]}
    )

# ── Webhook receiver: receive event, classify, act ────
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/webhook/{source}")
async def receive_webhook(source: str, payload: dict, background_tasks: BackgroundTasks):
    # Return 200 immediately (webhooks timeout at 30s)
    background_tasks.add_task(process_webhook, source, payload)
    return {"status": "received"}

async def process_webhook(source: str, payload: dict):
    try:
        if source == "github":
            if payload.get("action") == "opened" and "pull_request" in payload:
                pr = payload["pull_request"]
                review = await ai_review_pr(pr["diff_url"], pr["title"])
                await post_github_comment(pr["comments_url"], review)
        elif source == "stripe":
            event_type = payload.get("type", "")
            if event_type == "payment_intent.payment_failed":
                customer_id = payload["data"]["object"]["customer"]
                await notify_customer_of_failure(customer_id)
    except Exception as e:
        logger.error(f"Webhook processing failed: {e}", source=source)