Model Context Protocol (MCP)

from agentor.mcp import LiteMCP

mcp = LiteMCP(
    name="my-server",
    version="1.0.0",
    instructions="A simple MCP server"
)

@mcp.tool(description="Get weather for a given location")
def get_weather(location: str) -> str:
    return f"Weather in {location}: Sunny, 72°F"

mcp.serve()

mcp = LiteMCP(
    prefix="/mcp",                    # Endpoint prefix
    name="weather-server",            # Server name
    version="1.0.0",                  # Server version
    instructions="Weather services",  # Description
    website_url="https://example.com",
    icons=[...],                      # Optional icons
    dependencies=[...]                # FastAPI dependencies
)

mcp.serve(
    host="0.0.0.0",
    port=8000,
    enable_cors=True,  # Automatic CORS configuration
    reload=True,       # Auto-reload on changes
    log_level="debug"
)

@mcp.tool(
    name="search_docs",
    description="Search documentation",
    input_schema={  # Optional: custom JSON schema
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "limit": {"type": "integer"}
        },
        "required": ["query"]
    }
)
def search_docs(query: str, limit: int = 10) -> str:
    """Search documentation database"""
    return f"Found {limit} results for: {query}"

@mcp.tool()
def calculate(x: int, y: int, operation: str = "add") -> str:
    """Perform arithmetic operations"""
    if operation == "add":
        return str(x + y)
    return str(x - y)

{
  "type": "object",
  "properties": {
    "x": {"type": "integer"},
    "y": {"type": "integer"},
    "operation": {"type": "string"}
  },
  "required": ["x", "y"]
}

@mcp.resource(
    uri="config://settings",
    name="Application Settings",
    description="Current app configuration",
    mime_type="application/json"
)
def get_settings(uri: str) -> str:
    """Return current settings"""
    return '{"theme": "dark", "language": "en"}'

{
  "method": "resources/read",
  "params": {"uri": "config://settings"}
}

@mcp.prompt(
    name="code_review",
    description="Generate a code review prompt",
    arguments=[
        {
            "name": "language",
            "description": "Programming language",
            "required": True
        }
    ]
)
def code_review_prompt(language: str, style: str = "thorough") -> str:
    """Generate code review instructions"""
    return f"Review this {language} code with a {style} approach."

from agentor.mcp import get_token, get_context
from fastapi import Depends

@mcp.tool(description="Secure operation")
def secure_operation(location: str, ctx = Depends(get_context)) -> str:
    # Access headers
    user_agent = ctx.headers.get("user-agent")
    
    # Access cookies
    session_id = ctx.cookies.get("session_id")
    
    return f"Processing {location}"

from agentor.mcp import get_token

@mcp.tool()
def authenticated_tool(location: str) -> str:
    token = get_token()  # Extracts from "Bearer <token>"
    if token != "EXPECTED_SECRET":
        return "Not authorized"
    return f"Weather in {location}"

from fastapi import Depends

def get_db_connection():
    return {"connection": "active"}

@mcp.tool()
def query_database(
    query: str,
    db = Depends(get_db_connection)
) -> str:
    """Execute database query"""
    connection = db["connection"]
    return f"Executed on {connection}: {query}"

app = LiteMCP(name="my-server")

@app.tool()
def my_tool(param: str) -> str:
    return f"Result: {param}"

# Use with any ASGI server
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

gunicorn server:app -k uvicorn.workers.UvicornWorker

from fastapi import FastAPI
from agentor.mcp import LiteMCP

app = FastAPI()
mcp = LiteMCP(name="tools")

@mcp.tool()
def helper_tool(input: str) -> str:
    return f"Processed: {input}"

# Include MCP router
app.include_router(mcp.get_fastapi_router())

# Add your own routes
@app.get("/health")
def health():
    return {"status": "ok"}

@mcp.method("custom/action")
async def handle_custom(body: dict):
    params = body.get("params", {})
    # Process custom request
    return {"result": "success"}

from agentor.mcp import LiteMCP

app = LiteMCP(
    name="my-mcp-server",
    version="1.0.0",
    instructions="A simple MCP server example",
)

@app.tool(description="Get weather for a location")
def get_weather(location: str) -> str:
    """Get current weather for a location"""
    return f"🌤️ Weather in {location}: Sunny, 72°F"

@app.prompt(description="Generate a greeting")
def greeting(name: str, style: str = "formal") -> str:
    """Generate a personalized greeting"""
    if style == "formal":
        return f"Good day, {name}. How may I assist you today?"
    return f"Hey {name}! What's up?"

@app.resource(uri="config://settings", name="Settings", mime_type="application/json")
def get_settings(uri: str) -> str:
    """Get application settings"""
    return '{"theme": "dark", "language": "en"}'

if __name__ == "__main__":
    app.serve()

from agentor import Agentor, CelestoMCPHub
from agents.mcp import MCPServerStreamableHttp

# Connect to external MCP server
mcp_server = MCPServerStreamableHttp(
    name="External Tools",
    params={
        "url": "https://api.example.com/mcp",
        "headers": {"Authorization": "Bearer TOKEN"},
        "timeout": 10,
        "cache_tools_list": True
    }
)

agent = Agentor(
    name="Agent",
    tools=[mcp_server]
)

from agentor import CelestoMCPHub

async with CelestoMCPHub(api_key="YOUR_KEY") as mcp:
    agent = Agentor(
        name="Agent",
        tools=[mcp]
    )
    result = await agent.arun("Use Celesto tools")

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {"protocolVersion": "0.1.0"}
}

{
  "protocolVersion": "0.1.0",
  "capabilities": {
    "tools": {"listChanged": true},
    "resources": {"listChanged": true},
    "prompts": {"listChanged": true}
  },
  "serverInfo": {
    "name": "my-server",
    "version": "1.0.0"
  }
}