推荐支持MCP的六大AI框架

import asyncioimport shutilimport streamlit as stfrom agents import Agent, Runner, tracefrom agents.mcp import MCPServer, MCPServerStdioasync def query_git_repo(mcp_server: MCPServer, directory_path: str, query: str):    agent = Agent(        name="Assistant",        instructions=f"Answer questions about the localgit repository at {directory_path}, use that for repo_path",        mcp_servers=[mcp_server],    )    with st.spinner(f"Running query: {query}"):        result = await Runner.run(starting_agent=agent, input=query)        return result.final_outputasync def run_streamlit_app():    st.title("Local Git Repo Explorer")    st.write("This app allows you to query information about a local git repository.")    directory_path = st.text_input("Enter the path to the git repository:")    if directory_path:        # Common queries as buttons        col1, col2 = st.columns(2)        with col1:            if st.button("Most frequent contributor"):                query = "Who's the most frequent contributor?"                run_query(directory_path, query)        with col2:            if st.button("Last change summary"):                query = "Summarize the last change in the repository."                run_query(directory_path, query)        # Custom query        custom_query = st.text_input("Or enter your own query:")        if st.button("Run Custom Query") and custom_query:            run_query(directory_path, custom_query)def run_query(directory_path, query):    if not shutil.which("uvx"):        st.error("uvx is not installed. Please install it with `pip install uvx`.")        return    async def execute_query():        async with MCPServerStdio(            cache_tools_list=True,            params={                "command": "python",                 "args": [                    "-m",                     "mcp_server_git",                     "--repository",                     directory_path                ]            },        ) as server:            with trace(workflow_name="MCP Git Query"):                result = await query_git_repo(server, directory_path, query)                st.markdown("### Result")                st.write(result)    asyncio.run(execute_query())if __name__ == "__main__":    st.set_page_config(        page_title="Local Git Repo Explorer",        page_icon="?",        layout="centered"    )    # Change from async to synchronous implementation    # Since Streamlit doesn't work well with asyncio in the main thread    # Define a synchronous version of our app    def main_streamlit_app():        st.title("Local Git Repo Explorer")        st.write("This app allows you to query information about a Git repository.")        directory_path = st.text_input("Enter the path to the git repository:")        if directory_path:            # Common queries as buttons            col1, col2 = st.columns(2)            with col1:                if st.button("Most frequent contributor"):                    query = "Who's the most frequent contributor?"                    run_query(directory_path, query)            with col2:                if st.button("Last change summary"):                    query = "Summarize the last change in the repository."                    run_query(directory_path, query)            # Custom query            custom_query = st.text_input("Or enter your own query:")            if st.button("Run Custom Query") and custom_query:                run_query(directory_path, custom_query)    # Run the synchronous app    main_streamlit_app()

pip install streamlit openai-agents mcp-server-git

exportOPENAI_API_KEY=sk-...

# pip install chainlit
import chainlit as clfrom mcp import ClientSession
@cl.on_mcp_connectasyncdefon_mcp_connect(connection, session: ClientSession):"""Called when an MCP connection is established"""# Your connection initialization code here# This handler is required for MCP to work
@cl.on_mcp_disconnectasyncdefon_mcp_disconnect(name: str, session: ClientSession):"""Called when an MCP connection is terminated"""# Optional handler: Cleanup your code here

• 代表连接名称的唯一标识符。

• 客户端类型：你应当指定是使用 SSE 还是 stdio。使用 SSE 时，需要添加 URL 端点；而使用 stdio 时，则需要一个完整的命令（例如：npx your-tool-package）。下面是完整命令的示例：

npx -y linear-mcp-server --tools=all --api-key=lin_api_your_linear_API_Key

代码链接：https://github.com/Chainlit/cookbook/tree/main/mcp-linear

# Define server parameters    airbnb_server_params = StdioServerParameters(        command="npx",        args=["-y", "@openbnb/mcp-server-airbnb", "--ignore-robots-txt"],        env=env,    )
    maps_server_params = StdioServerParameters(        command="npx", args=["-y", "@modelcontextprotocol/server-google-maps"], env=env    )
# Use contextlib.AsyncExitStack to manage multiple async context managersasyncwith contextlib.AsyncExitStack() as stack:# Create stdio clients for each server        airbnb_client, _ = await stack.enter_async_context(stdio_client(airbnb_server_params))        maps_client, _ = await stack.enter_async_context(stdio_client(maps_server_params))
# Create all agents        airbnb_agent = Agent(            name="Airbnb",            role="Airbnb Agent",            model=OpenAIChat("gpt-4o"),            tools=[airbnb_client],            instructions=dedent("""\                You are an agent that can find Airbnb listings for a given location.\            """),            add_datetime_to_instructions=True,        )