// Kailash MCP (Rust) — server, client, tools, resources, auth, transports. For AI agent integration.
[HINT] Download the complete skill directory including SKILL.md and all related files
| name | kailash-mcp |
| description | Kailash MCP (Rust) — server, client, tools, resources, auth, transports. For AI agent integration. |
Production-ready MCP server implementation built into Kailash Core SDK for seamless AI agent integration.
Kailash's MCP module provides:
from kailash.mcp import McpApplication, prompt_argument
app = McpApplication("my-server", "1.0")
@app.tool("search", "Search the web")
def search(params):
return f"Results for {params['query']}"
@app.resource(uri="config://settings", name="Settings")
def get_settings(uri: str) -> str:
return '{"theme": "dark"}'
@app.prompt("summarize", description="Summarize text")
def summarize_prompt(arguments):
return [{"role": "user", "content": f"Please summarize: {arguments['text']}"}]
from kailash import McpServer
# Create MCP server -- name and version are required
server = McpServer("my-server", version="1.0.0")
# Register workflow as MCP tool
def summarize_handler(args: dict) -> dict:
"""Summarize the given text."""
text = args.get("text", "")
return {"summary": text[:100]}
server.register_tool(
"summarize",
"Summarize the given text",
summarize_handler,
schema={"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]},
)
# Note: McpServer does not have a run() method.
# To serve MCP tools over a network, use Nexus:
# from kailash.nexus import NexusApp
# app = NexusApp(config=NexusConfig(enable_mcp=True))
print(f"Tools registered: {server.tool_count()}")
The Model Context Protocol enables AI agents to:
MCP supports multiple transport mechanisms:
Tools are type-safe functions exposed to AI agents:
Resources expose data to AI agents:
Use MCP when you need to:
import kailash
reg = kailash.NodeRegistry()
server = McpServer("workflow-server", version="1.0.0")
def process_handler(args: dict) -> dict:
builder = kailash.WorkflowBuilder()
# Build workflow
results = rt.execute(builder.build(reg))
return results["results"]["output"]["result"]
server.register_tool("process_data", "Process data", process_handler)
from kailash.nexus import NexusApp, NexusConfig
# Nexus automatically creates MCP channel
app = NexusApp(config=NexusConfig(port=3000, enable_mcp=True))
@app.handler(name="summarize", description="Summarize text")
async def summarize(text: str) -> dict:
return {"summary": text[:100]}
app.start() # Includes MCP server
import kailash
server = McpServer("db-server", version="1.0.0")
df = kailash.DataFlow(...)
server.register_resource(
uri="data://users",
name="Users",
content="User data from database",
description="Expose database users via MCP resource",
)
import kailash
server = McpServer("agent-server", version="1.0.0")
def analyze_handler(args: dict) -> dict:
from kailash.kaizen import BaseAgent
# Use a custom BaseAgent subclass here
return {"output": f"Analyzed: {args.get('text', '')}"}
server.register_tool("analyze", "Analyze text", analyze_handler)
| Transport | Use Case | Pros | Cons |
|---|---|---|---|
| stdio | Local tools, CLI | Simple, reliable | Local only |
| SSE | Web apps | Real-time updates | Complex setup |
| HTTP | APIs, services | Standard protocol | No streaming |
For MCP-specific questions, invoke:
mcp-specialist - MCP server implementationtesting-specialist - MCP testing strategies skill - MCP integration architecture