支持 MCP 的七大 AI 框架

import asyncioimport shutilimport streamlit as stfrom agents importAgent,Runner, tracefrom agents.mcp importMCPServer,MCPServerStdio
# 执行 Git 查询任务async 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_output
# 查询封装def run_query(directory_path, query):ifnot shutil.which("uvx"):        st.error("uvx 未安装，请运行 `pip install uvx` 进行安装。")return
    async def execute_query():        async withMCPServerStdio(            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("### 查询结果")                st.write(result)
    asyncio.run(execute_query())
# 主页面逻辑（同步版本）def main_streamlit_app():    st.title("本地 Git 仓库查询助手")    st.write("可通过自然语言查询本地 Git 仓库的信息。")
    directory_path = st.text_input("请输入 Git 仓库路径：")
if directory_path:        col1, col2 = st.columns(2)with col1:if st.button("最频繁的贡献者"):                run_query(directory_path, query)
with col2:if st.button("最近提交摘要"):                query ="Summarize the last change in the repository."                run_query(directory_path, query)
        custom_query = st.text_input("或输入你自己的查询内容：")if st.button("运行自定义查询")and custom_query:            run_query(directory_path, custom_query)
if __name__ =="__main__":    st.set_page_config(page_title="Local Git Repo Explorer", page_icon="?", layout="centered")    main_streamlit_app()

import streamlit as stfrom praisonaiagents importAgent, MCP
st.title("? Airbnb Booking Assistant")
# 创建代理（MCP 工具通过 npx 启动 Airbnb 服务）@st.cache_resourcedef get_agent():returnAgent(        instructions="""You help book apartments on Airbnb.""",        llm="gpt-4o-mini",# 使用 GPT-4o-mini 模型        tools=MCP("npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"))
# 初始化聊天记录if"messages"notin st.session_state:    st.session_state.messages =[]
# 显示聊天记录for message in st.session_state.messages:with st.chat_message(message["role"]):        st.markdown(message["content"])
# 用户输入表单with st.form("booking_form"):    st.subheader("请输入您的预订信息")
    destination = st.text_input("目的地：","巴黎")
    col1, col2 = st.columns(2)with col1:        check_in = st.date_input("入住日期")with col2:        check_out = st.date_input("退房日期")
    adults = st.number_input("成人人数：", min_value=1, max_value=10, value=2)
    submitted = st.form_submit_button("搜索住宿")
if submitted:        search_agent = get_agent()
# 构建查询内容        query = f"I want to book an apartment in {destination} from {check_in.strftime('%m/%d/%Y')} to {check_out.strftime('%m/%d/%Y')} for {adults} adults"
# 添加用户输入到聊天记录        st.session_state.messages.append({"role":"user","content": query})
# 显示用户输入with st.chat_message("user"):            st.markdown(query)
# 获取代理响应with st.chat_message("assistant"):with st.spinner("正在查找住宿信息..."):                response = search_agent.start(query)                st.markdown(response)
# 添加助手响应到聊天记录        st.session_state.messages.append({"role":"assistant","content": response})
# 跟进问题输入if st.session_state.messages:    prompt = st.chat_input("对住宿还有其他问题？可以继续提问")if prompt:        search_agent = get_agent()
# 添加用户输入到聊天记录        st.session_state.messages.append({"role":"user","content": prompt})
# 显示用户输入with st.chat_message("user"):            st.markdown(prompt)
# 获取代理响应with st.chat_message("assistant"):with st.spinner("思考中..."):                response = search_agent.start(prompt)                st.markdown(response)
# 添加助手响应到聊天记录        st.session_state.messages.append({"role":"assistant","content": response})

# Copyright (C) 2024 Andrew Wason# SPDX-License-Identifier: MIT
import asyncioimport pathlibimport sysimport typing as t
from langchain_core.messages importAIMessage,BaseMessage,HumanMessagefrom langchain_core.output_parsers importStrOutputParserfrom langchain_core.tools importBaseToolfrom langchain_groq importChatGroqfrom mcp importClientSession,StdioServerParametersfrom mcp.client.stdio import stdio_client
from langchain_mcp importMCPToolkit
# 主推理执行函数：运行模型与工具交互async def run(tools: list[BaseTool], prompt: str)-> str:    model =ChatGroq(model_name="llama-3.1-8b-instant", stop_sequences=None)# 需设置 GROQ_API_KEY    tools_map ={tool.name: tool for tool in tools}    tools_model = model.bind_tools(tools)    messages: list[BaseMessage]=[HumanMessage(prompt)]    ai_message = t.cast(AIMessage, await tools_model.ainvoke(messages))    messages.append(ai_message)
# 执行工具调用for tool_call in ai_message.tool_calls:        selected_tool = tools_map[tool_call["name"].lower()]        tool_msg = await selected_tool.ainvoke(tool_call)        messages.append(tool_msg)
return await (tools_model |StrOutputParser()).ainvoke(messages)
# 主程序入口async def main(prompt: str)->None:# 配置 MCP 文件系统工具（通过 npx 启动）    server_params =StdioServerParameters(        command="npx",        args=["-y","@modelcontextprotocol/server-filesystem", str(pathlib.Path(__file__).parent.parent)],)
    async with stdio_client(server_params)as(read, write):        async withClientSession(read, write)as session:            toolkit =MCPToolkit(session=session)            await toolkit.initialize()            response = await run(toolkit.get_tools(), prompt)print(response)
# 默认命令：读取并总结 README.md 文件if __name__ =="__main__":    prompt = sys.argv[1]if len(sys.argv)>1else"Read and summarize the file ./readme.md"    asyncio.run(main(prompt))

# pip install chainlit
import chainlit as clfrom mcp importClientSession
@cl.on_mcp_connectasync def on_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_disconnectasync def on_mcp_disconnect(name: str, session:ClientSession):"""Called when an MCP connection is terminated"""# Optional handler: Cleanup your code here

# 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 managers    async with 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,)

import osfrom dotenv import load_dotenvfrom upsonic importTask,Agent,Directfrom upsonic.client.tools importSearch# Adding Search as a fallback tool
# Load environment variables from .env fileload_dotenv()
# Get the OpenAI API key from environment variablesopenai_api_key = os.getenv("OPENAI_API_KEY")ifnot openai_api_key:raiseValueError("OPENAI_API_KEY not found in .env file")
# Set your OpenAI API key for the sessionos.environ["OPENAI_API_KEY"]= openai_api_key
# Define the HackerNews MCP tool# Using the correct MCP setup for HackerNews based on Upsonic documentationclassHackerNewsMCP:    command ="uvx"    args =["mcp-hn"]# No environment variables are needed for this MCP
# Create a task to analyze the latest HackerNews stories# Adding Search as a fallback in case HackerNews MCP failstask =Task("Analyze the top 5 HackerNews stories for today. Provide a brief summary of each story, ""identify any common themes or trends, and highlight which stories might be most relevant ""for someone interested in AI and software development.",    tools=[HackerNewsMCP,Search]# Include both HackerNews MCP and Search tools)
# Create an agent specialized in tech news analysisagent =Agent("Tech News Analyst",    company_url="https://news.ycombinator.com/",    company_objective="To provide insightful analysis of tech industry news and trends")
# Execute the task with the agent and print the resultsprint("Analyzing HackerNews stories...")agent.print_do(task)
# Alternatively, you can use a Direct LLM call if the task is straightforward# print("Direct analysis of HackerNews stories...")# Direct.print_do(task)
# If you want to access the response programmatically:# agent.do(task)# result = task.response# print(result)

import{MCPConfiguration}from"@mastra/mcp";import{Agent}from"@mastra/core/agent";import{ openai }from"@ai-sdk/openai";
const mcp =newMCPConfiguration({  servers:{    stockPrice:{      command:"npx",      args:["tsx","stock-price.ts"],      env:{        API_KEY:"your-api-key",},},    weather:{      url:new URL("http://localhost:8080/sse"),},},});
// Create an agent with access to all toolsconst agent =newAgent({  name:"Multi-tool Agent",  instructions:"You have access to multiple tool servers.",  model: openai("gpt-4"),  tools: await mcp.getTools(),});