from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="my-mcp-server",
    stateless_http=True,
    json_response=True
)

@mcp.tool(name="my_tool", description="Tool description")
def my_tool(param: str) -> str:
    return f"Result: {param}"

# Create HTTP app
streamable_http_app = mcp.streamable_http_app()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("server:streamable_http_app", host="0.0.0.0", port=8001)

import asyncio
from agents import Agent, Runner, OpenAIChatCompletionsModel
from agents.mcp import MCPServerStreamableHttp, MCPServerStreamableHttpParams
from openai import AsyncOpenAI

MCP_SERVER_URL = "http://localhost:8001/mcp/"

async def main():
    client = AsyncOpenAI(api_key="your-api-key")
    mcp_params = MCPServerStreamableHttpParams(url=MCP_SERVER_URL)

    async with MCPServerStreamableHttp(
        params=mcp_params,
        name="MyMCPClient"
    ) as mcp_server:
        agent = Agent(
            name="MyAgent",
            instructions="You are a helpful assistant.",
            mcp_servers=[mcp_server],
            model=OpenAIChatCompletionsModel(model="gpt-4o", openai_client=client)
        )

        result = await Runner.run(agent, "Your query here")
        print(result.final_output)

asyncio.run(main())

MCPServerStreamableHttp(
    params=mcp_params,
    cache_tools_list=True  # Enable caching
)

from agents.mcp import create_static_tool_filter

tool_filter = create_static_tool_filter(
    allowed_tool_names=["allowed_tool"],
    blocked_tool_names=["blocked_tool"]
)

MCPServerStreamableHttp(params=mcp_params, tool_filter=tool_filter)

from agents.mcp import ToolFilterContext

def context_aware_filter(context: ToolFilterContext, tool) -> bool:
    return context.agent.name == "MyAgent" and tool.name == "my_tool"

MCPServerStreamableHttp(params=mcp_params, tool_filter=context_aware_filter)

from contextlib import AsyncExitStack

async def connect_multiple_servers(urls: list):
    mcp_servers = []
    async with AsyncExitStack() as stack:
        for url in urls:
            params = MCPServerStreamableHttpParams(url=url)
            client = await stack.enter_async_context(
                MCPServerStreamableHttp(params=params, name=f"Client_{url}")
            )
            mcp_servers.append(client)

        agent = Agent(
            name="MultiServerAgent",
            mcp_servers=mcp_servers,
            model=model
        )
        return agent

@mcp.tool(name="tool_name", description="Tool description")
def my_tool(param: str, optional: str = "default") -> str:
    """Docstring becomes parameter description."""
    return f"Result: {param}"

@mcp.resource("docs://{doc_id}", mime_type="text/plain")
def get_document(doc_id: str) -> str:
    return documents[doc_id]

from pydantic import Field

@mcp.prompt(name="my_prompt", description="Prompt description")
def generate_instructions(focus: str = Field(description="Focus area")) -> str:
    return f"Instructions focused on {focus}..."

from agents import input_guardrail, GuardrailFunctionOutput

@input_guardrail
async def validate_input(ctx, agent, input):
    if "forbidden" in input:
        return GuardrailFunctionOutput(tripwire_triggered=True)
    return GuardrailFunctionOutput(tripwire_triggered=False)

from agents import output_guardrail

@output_guardrail
async def validate_output(ctx, agent, output):
    # Validate final output
    return GuardrailFunctionOutput(tripwire_triggered=False)

from agents import handoff

specialist = Agent(name="Specialist", instructions="...")

main_agent = Agent(
    name="Main",
    handoffs=[specialist]  # Direct handoff
)

from agents import handoff

custom_handoff = handoff(
    agent=specialist,
    tool_name_override="escalate_to_specialist",
    tool_description_override="Transfer to specialist for complex issues"
)

from dataclasses import dataclass
from agents import RunContextWrapper

@dataclass
class MyContext:
    user_id: str
    session_id: str

async def main():
    ctx = MyContext(user_id="123", session_id="abc")
    result = await Runner.run(agent, "query", context=ctx)

from agents import trace

with trace("My Workflow"):
    result1 = await Runner.run(agent, "First query")
    result2 = await Runner.run(agent, "Second query")

try:
    result = await Runner.run(agent, query)
except MaxTurnsExceeded:
    # Handle max turns exceeded
    pass
except InputGuardrailTripwireTriggered:
    # Handle guardrail triggered
    pass