npx claudepluginhub rehan-ul-haq/my-skills --plugin my-skillsThis skill uses the workspace's default tool permissions.
Build MCP servers and integrate them with OpenAI ChatKit for conversational AI applications.
Guides creating MCP servers enabling LLMs to interact with external services via tools. Use when building integrations in Python (FastMCP) or Node/TypeScript (MCP SDK).
Builds MCP servers in Python (FastMCP) or Node/TypeScript (MCP SDK) to expose third-party APIs as LLM tools. Handles scaffolding, adding tools, evaluations, and tool interface design.
Guides developers building MCP servers for Claude: interrogates use case, selects deployment (remote HTTP, MCPB, stdio), tool patterns, and hands off to specialized skills.
Share bugs, ideas, or general feedback.
Build MCP servers and integrate them with OpenAI ChatKit for conversational AI applications.
┌─────────────┐ ┌─────────────────┐ ┌─────────────┐
│ ChatKit │────▶│ Dispatcher │────▶│ MCP Server │
│ Frontend │ │ Agent (SDK) │ │ (FastMCP) │
└─────────────┘ └─────────────────┘ └─────────────┘
│ │
▼ ▼
MCPServerStreamableHttp Database/APIs
from fastmcp import FastMCP
from fastapi import FastAPI
mcp = FastMCP("My Tools")
@mcp.tool()
async def my_tool(user_id: str, param: str) -> dict:
"""Tool description for the agent."""
return {"result": f"Processed {param} for {user_id}"}
# Mount as FastAPI app
mcp_asgi = mcp.http_app(transport="streamable-http", path="/")
app = FastAPI(lifespan=mcp_asgi.lifespan)
app.mount("/mcp", mcp_asgi)
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp
mcp_server = MCPServerStreamableHttp(
name="My MCP",
params={"url": "http://localhost:8001/mcp", "timeout": 60},
client_session_timeout_seconds=30,
cache_tools_list=True,
)
async with mcp_server:
agent = Agent(
name="Assistant",
instructions=dynamic_instructions, # Function for user_id injection
model="gpt-4o",
mcp_servers=[mcp_server],
)
result = await Runner.run(agent, "Hello", context=my_context)
from chatkit.server import ChatKitServer
from chatkit.agents import stream_agent_response
class MyChatKitServer(ChatKitServer[dict]):
async def respond(self, thread, input, context):
result = Runner.run_streamed(agent, input, context=agent_ctx)
async for event in stream_agent_response(agent_ctx, result):
yield event
MCP tools need user_id for data isolation. Use dynamic instructions:
def get_dynamic_instructions(context_wrapper, agent) -> str:
user_id = "anonymous"
ctx = context_wrapper.context
if hasattr(ctx, "user_id"):
user_id = ctx.user_id
elif hasattr(ctx, "request_context"):
user_id = ctx.request_context.get("user_id", "anonymous")
return BASE_PROMPT.format(user_id=user_id)
Default MCP timeout is 5 seconds - increase for database operations:
MCPServerStreamableHttp(
params={"url": url, "timeout": 60},
client_session_timeout_seconds=30, # MCP session read timeout
)
from chatkit.types import ThreadItemAddedEvent, AssistantMessageItem
except Exception as e:
error_item = AssistantMessageItem(
id=store.generate_item_id("message", thread, context),
content=[AssistantMessageContent(type="output_text", text="Error occurred")],
)
yield ThreadItemAddedEvent(type="item.added", item=error_item)
| Issue | Solution |
|---|---|
| Timeout after 5s | Increase client_session_timeout_seconds |
| Tools not discovered | Check MCP URL (no trailing slash) |
| user_id not passed | Use dynamic instructions pattern |
| Closure bug in tool wrappers | Use SDK's built-in mcp_servers parameter |
| ThreadItemCreatedEvent not found | Use ThreadItemAddedEvent instead |
MCP_SERVER_URL=http://localhost:8001/mcp # No trailing slash
OPENAI_API_KEY=your_key