用Python和TypeScript框架,基于MCP服务器为LLM赋能,打造AI应用
MCP支持的AI框架
AI代理工具包为开发者开放了各种API,让AI解决方案具备执行任务的工具,确保能给出准确结果,提升用户满意度。然而,把这些工具集成到AI应用里并进行管理,过程往往很繁琐。本文将为你介绍一种行业标准方法——利用模型上下文协议(MCP),为大语言模型(LLM)和代理提供上下文信息。
LLM上下文提供方法和规范
默认情况下,如果不给LLM和AI聊天机器人设定合适的上下文,它们就无法获取实时信息、执行代码、调用外部工具及API,甚至不能代用户使用网页浏览器。开发者可借助以下方法,突破LLM和代理的这些局限。
- Composio:Composio有用于集成人工智能代理和LLM的规范及工具包库。除了现成的工具包库,他们最近还推出了Composio MCP,开发者借此能为集成开发环境(IDE)连接100多个MCP服务器。从上面的链接查看Composio MCP工具类别,就能在Cursor、Claude和Windsurf等支持MCP的IDE中,把多个应用连接到自己的项目里。
- Agents.json:这是基于OpenAI标准构建的规范,旨在保障AI代理之间交互顺畅,方便其访问API和外部工具。尽管Agent.json很出色,但不像MCP那样被广泛应用和采用。想深入了解并上手的话,可以参考其GitHub仓库。
- MCP:MCP为开发者提供了向LLM和AI助手提供上下文数据以解决问题的最佳方式。例如,你可以搭建一个MCP文档服务器,让IDE和代理框架能全面访问文档内容,就像使用llms.txt文件一样。
什么是MCP?
不妨把MCP看作LLM的第三次重大变革。在LLM发展的第一阶段,若能在训练数据里找到对应查询,LLM就能准确回答用户的提示信息。但这个阶段的LLM无法处理训练数据之外的提示,因为它们无法访问外部工具。到了第二阶段,虽然给LLM接入了额外的上下文(工具),但使用起来并不直观。不过,这些工具能帮助LLM更精准地预测和理解用户意图并做出回应。而第三次变革,依然是围绕LLM和工具展开,只是构建了一套更完善的基础设施,让它们能便捷地访问外部应用,并且更易于维护。
构建人工智能服务时,在企业场景下用于回复客户支持工单的AI助手应用,其数据可能存于云端。MCP是Anthropic推出的开源协议,借助它,能把企业数据接入AI系统。
MCP提供了一种标准化途径,可将内容存储库(如GitHub、Notion)、开发环境、网页及商业工具与辅助AI技术连接起来。MCP一个热门且应用越来越广泛的场景是AI辅助编程。Cursor和Windsurf等开发环境及工具,通过数百个MCP集成,让开发者在开发过程中能与外部应用连接并交互。
注意:本文聚焦于使用Python和TypeScript构建AI助手和代理系统时如何实现MCP,而非基于IDE的MCP集成。
MCP是如何工作的
在LLM和代理的应用场景中,MCP能帮助它们针对内置知识以外的用户提问,给出有价值的回复。比方说,让ChatGPT给特定的Slack频道发消息、查看你的日历日程并安排与队友的会议,它肯定无法满足你的要求,因为它无法访问这些应用。但要是集成了MCP,就能让这类助手给出有用的反馈。
开发者最先好奇的往往是MCP的工作原理。在MCP的基础运行流程里,用户向代理发出查询请求,代理随后判断调用哪个MCP服务器和工具,获取操作所需的相关信息,最后利用特定工具返回的数据答复用户。
为什么要为AI代理和基于LLM的应用程序采用MCP?
MCP正逐渐成为一项行业标准,助力开发者打造能与外部应用高效通信的AI系统。微软近期宣布在Copilot Studio中集成MCP,简化了AI应用和代理访问工具的流程。OpenAI也宣称其包括Agents SDK和ChatGPT桌面应用等多款产品都支持MCP。直接给工具配备AI助手并非不可行,只是对于由多个执行不同任务(如读取和回复邮件、网页数据抓取、财务分析、获取实时天气信息等)的多代理构成的AI代理系统而言,管理起来会变得异常复杂。
具有工具集成的AI代理
在上图中,三个外部工具连接到LLM。如果工具数量增加到100多个,管理和保障它们的安全会让人焦头烂额。改进方法是通过MCP注册表访问同样多甚至更多的工具,下一部分会详细说明。
具有MCP集成的AI代理
在此图中,我们整合了代理系统所需的工具,并通过MCP服务器访问它们,以提供更有凝聚力的用户体验。MCP方法使通过中心位置保护和管理这些工具变得更加容易。
使用MCP相对于传统工具包集成的优势
与传统的将工具集成到AI代理的方式相比,MCP有诸多显著优势:
- 架构灵活:与AI代理预构建的工具规范不同,MCP的架构简洁且灵活,与工具和API交互更顺畅。
- 管理更优:MCP通过标准化接口,让AI模型能便捷地访问工具,有效解决了LLM与第三方系统通信不畅的问题。
- 适用性广:MCP工具既能满足单用户需求,也适用于团队协作场景,弥补了独立工具的局限性。
- 社区支持:MCP拥有丰富的开源服务器资源和开发者生态系统,在众多应用场景中都得到了开发者社区的广泛认可。
- 认证安全:MCP内置强大的认证和权限系统,严格管控工具访问权限。比如使用Composio提供的MCP工具时,能用Google表格或Gmail对用户进行身份验证。
- 工具查找便捷:与传统安装、配置、集成AI聊天机器人工具的繁琐流程不同,MCP能让开发者更轻松地搜索和找到外部工具。
- 易于扩展:MCP可轻松扩展,满足大量用户和应用程序的使用需求。
- 行业标准规范:虽然可以通过安装硬编码工具为AI应用提供上下文,但MCP提供了更规范的行业标准方案。
MCP服务器的种类
Anthropic的MCP规范定义了两种服务器类型,方便为代理和AI项目添加工具:
- 服务器发送事件(SSE):通过HTTP协议连接远程服务。
- STDIO:支持通过标准输入/输出执行本地命令和通信。
构建AI应用时选用的框架,会提供连接这些服务器所需的类。
访问MCP注册表/服务器的生态系统
有不少托管MCP工具的开源库,能增强LLM和代理的能力,确保它们给出的回复可靠。这些工具库被称为注册表,里面汇聚了各类精选服务。借助它们的工具,能把AI应用连接到以下注册表。此外,还能选择不同类型的服务器,像uvx,它基于Python开发,使用时无需安装;也有基于Docker运行MCP工具的方案,以及需要安装Node.js的npx服务器。
- GitHub上的MCP服务器:这是社区构建的服务器集合,包含丰富的MCP资源。
- Glama Registry:面向开发者的生产就绪型开源MCP服务器。
- Smithery Registry:借助Smithery,开发者能访问2000多个MCP服务器,大幅拓展AI代理和LLM的功能。
- OpenTools:OpenTools为MCP工具使用提供生成性API。利用它,开发者能获取数百个现成的MCP工具用于AI项目开发。借助OpenTools API,还能拓展LLM的网页搜索、实时位置数据获取和网页数据抓取等功能。该API支持Curl、Python和TypeScript。想使用API的话,可以查看OpenTools快速入门指南。
from openai import OpenAI
client = OpenAI(
base_url="https://api.opentools.com",
api_key="<OPENTOOLS_API_KEY>"
)
completion = client.chat.completions.create(
model="anthropic/claude-3.7-sonnet",
messages=[
{ "role": "user", "content": "Compare specs of top 5 EVs on caranddriver.com" }
],
tools=[{ "type": "mcp", "ref": "firecrawl" }]
)
- PulseMCP Registry:使用PulseMCP,能浏览适合AI项目的托管MCP工具和应用案例。关注PulseMCP News,还能了解近期热门的MCP服务器和应用。
- mcp.run:通过这个注册表,开发者能访问数百个用于商业用途的MCP应用。
- Composio Registry:Composio基于SSE的MCP服务器,能轻松将工具与不同AI框架集成,便于开发应用程序。
- guMCP:Gumloop的guMCP提供免费、开源且全托管的MCP服务器,可与任何AI应用无缝集成。
将MCP添加到LLM和代理的7大客户端框架
虽说MCP如今热度很高,开发人员社区都在讨论,但要选对与AI应用和代理集成的MCP客户端框架并非易事。经过研究,下面为大家介绍几款领先的、适用于基于Python和TypeScript的代理工作流及AI助手的MCP客户端平台。
- 使用OpenAI代理SDK构建Git MCP代理:利用OpenAI Agents SDK构建代理时,借助SDK里的MCPServerStdio和MCPServerSse类,就能连接到社区构建的MCP服务器。下面的MCP代理代码实现了访问本地Git仓库根目录,并能响应用户对仓库相关的查询。
import asyncio
import shutil
import streamlit as st
from agents import Agent, Runner, trace
from agents.mcp import MCPServer, MCPServerStdio
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
async 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:
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 = 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"
)
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:
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 = st.text_input("Or enter your own query:")
if st.button("Run Custom Query") and custom_query:
run_query(directory_path, custom_query)
main_streamlit_app()
上面的代码将Streamlit与OpenAI MCP代理集成,这样就能用Git MCP服务器与本地Git仓库交互了。运行这个示例前,需要安装以下依赖:
pip install streamlit openai-agents mcp-server-git
然后,通过export OPENAI_API_KEY=sk-....
导出OpenAI API密钥。
你还能在GitHub上探索更多OpenAI MCP的示例。使用Agents SDK的MCP集成有个突出优势,OpenAI的仪表板上内置了MCP代理监控系统,能自动记录代理的MCP操作,比如工具列表、POST响应以及函数调用数据等信息。
2. 利用Praison AI构建MCP人工智能代理
Praison AI是一个基于Python的人工智能框架,用于组建代理团队。它提供了一种极为简便的方式,仅需一行代码,就能像为代理配备传统工具那样,将MCP服务器工具融入代理工作流程。
下面的示例通过Streamlit用户界面,把Airbnb的MCP服务器与Praison AI代理进行集成,以此帮助用户在指定地点寻找公寓。若要使用Praison AI创建你的首个MCP代理,需安装以下内容:
pip install praisonaiagents mcp streamlit
接下来,导出你的OpenAI API密钥:export OPENAI_API_KEY='sk-proj-qZIGbi....'
创建一个Python文件,比如streamlit_praison_airbnb_mcp_agent.py
,并将以下代码写入其中:
import streamlit as st
from praisonaiagents import Agent, MCP
st.title("🏠 Airbnb预订助手")
@st.cache_resource
def get_agent():
return Agent(
instructions="""你帮助在爱彼迎上预订公寓。""",
llm="gpt-4o-mini",
tools = MCP("npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt")
)
if "messages" not in 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"我想在{destination}预订一间公寓,从{check_in.strftime('%m/%d/%Y')}入住至{check_out.strftime('%m/%d/%Y')},共{adults}位成人"
st.session_state.messages.append({"role": "用户", "content": query})
with st.chat_message("用户"):
st.markdown(query)
with st.chat_message("助手"):
with st.spinner("正在搜索住宿..."):
response = search_agent.start(query)
st.markdown(response)
st.session_state.messages.append({"role": "助手", "content": response})
if st.session_state.messages:
prompt = st.chat_input("询问关于住宿的后续问题")
if prompt:
search_agent = get_agent()
st.session_state.messages.append({"role": "用户", "content": prompt})
with st.chat_message("用户"):
st.markdown(prompt)
with st.chat_message("助手"):
with st.spinner("思考中..."):
response = search_agent.start(prompt)
st.markdown(response)
st.session_state.messages.append({"role": "助手", "content": response})
运行这段示例代码,它会调用所需的Airbnb MCP工具,帮你在特定地点寻找住所。
搭载Praison AI的MCP人工智能代理
你会发现,它仅用一行代码tools = MCP("npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt")
就为代理添加了MCP支持。这里的npx
是用于启动MCP服务器的命令,-y
是传递给该命令的命令行参数。想要了解更多相关信息,可参考OpenAI Agents SDK文档中的MCP服务器部分。
3. 将MCP用于LangChain AI应用程序
LangChain支持调用MCP工具,这一特性允许开发者通过设置Python函数,访问不同的MCP服务器,并获取工具以在AI项目中执行任务。下面的示例代码展示了如何连接到一个安全的MCP文件系统服务器,使大语言模型(LLM)能够准确回答与你提供的任何文件相关的问题。
import asyncio
import pathlib
import sys
import typing as t
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import BaseTool
from langchain_groq import ChatGroq
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp import MCPToolkit
async def run(tools: list[BaseTool], prompt: str) -> str:
model = ChatGroq(model_name="llama-3.1-8b-instant", stop_sequences=None)
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:
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 with ClientSession(read, write) as session:
toolkit = MCPToolkit(session=session)
await toolkit.initialize()
response = await run(toolkit.get_tools(), prompt)
print(response)
if __name__ == "__main__":
prompt = sys.argv[1] if len(sys.argv) > 1 else "Read and summarize the file ./readme.md"
asyncio.run(main(prompt))
在运行这段Python脚本之前,你需要安装所需的依赖项:langchain - core
、langchain - groq
和 langchain - mcp
。
pip install langchain-core langchain-groq langchain-mcp
上述MCP配置使用了npx服务器类型,因此你还需要安装 server - filesystem
包。
pm install -g @modelcontextprotocol/server-filesystem
注意:本示例取自LangChain的GitHub仓库。
4. 将MCP用于Chainlit AI应用程序
Chainlit是一个用Python构建AI应用程序的平台。它内置了对MCP服务器的支持,所以你可以配置应用程序,发现可用的MCP工具,并将工具调用集成到应用程序流程中,以获得更好的结果。你可以将Chainlit应用程序与基于服务器发送事件(SSE)和命令行(stdio)的服务集成。在下面的示例中,我们将把一个Chainlit应用程序连接到Linear MCP服务器,使该应用程序能够管理Linear中的问题、项目和团队。你可以使用本示例中提供的Linear工具来创建、更新、搜索和获取用户问题,或者为问题添加评论。
配置你的Chainlit应用程序以连接到MCP服务器
将Chainlit应用程序连接到MCP服务器以访问工具,主要涉及两个步骤。
- 注册MCP连接:在这一步中,你需要实现Chainlit的
on_mcp_connect
异步函数,以建立成功的连接。你也可以实现on_mcp_disconnect
函数来处理连接断开时的清理工作。
import chainlit as cl
from mcp import ClientSession
@cl.on_mcp_connect
async def on_mcp_connect(connection, session: ClientSession):
"""当MCP连接建立时调用"""
@cl.on_mcp_disconnect
async def on_mcp_disconnect(name: str, session: ClientSession):
"""当MCP连接终止时调用"""
- 配置MCP客户端(Chainlit、LangChain、Mastra):为了让MCP服务器与Chainlit应用程序协同工作,客户端需要通过Chainlit的用户界面提供连接详细信息。此配置包含以下内容:
- 唯一标识符:一个代表连接名称的唯一标识。
- 客户端类型:你需要指定是使用sse还是stdio。如果使用sse,需要添加一个URL端点;使用stdio时,则需要一个完整的命令(例如,
npx your - tool - package
或uvx your - tool - package
)。下面是一个完整的命令示例:
npx -y linear-mcp-server --tools=all --api-key=lin_api_your_linear_API_Key
建立MCP服务器连接后,你可以使用MCP会话来执行工具。最后,通过工具调用,将MCP工具与Chainlit应用程序的模型/代理无缝集成。你可以从GitHub上Chainlit的示例应用程序中找到此Linear MCP集成的完整源代码。
当你从Chainlit的GitHub仓库获取上述源代码并运行,然后通过Chainlit界面设置 npx -y linear - mcp - server -- tools = all -- api - key = lin_api_your_linear_API_Key
,你将能够创建和更新Linear中的问题/项目。
构建连接到MCP服务器的Chainlit应用程序
5. 为Agno AI代理集成MCP
Agno是一个用于构建复杂代理工作流的Python框架。它因其简单性、易用性以及与MCP服务器的无缝集成而广受欢迎。本节中的示例MCP实现,与由四个独立的贡献代理(如Airbnb、Google Maps、网络搜索和天气MCP代理)组成的多代理团队进行了集成。Agno的多个代理相互协作,提供关于特定地点旅行的信息。
先决条件
为了测试本节中的Agno MCP实现示例:
- 安装Agno、DuckDuckGo和Exa:
pip install -U openai agno duckduckgo - search exa - py
。 - 获取
GOOGLE_MAPS_API_KEY
,并将其添加到项目的.env
文件中。 - 获取一个
APIFY_TOKEN
,并添加到你的.env
文件中。 - 验证Google地址API。
配置Agno MCP代理团队
在这一步中,你需要定义MCP服务器参数,并使用 AsyncExitStack
管理多个上下文管理器。然后,创建代理并运行它们。
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
)
async with contextlib.AsyncExitStack() as stack:
airbnb_client, _ = await stack.enter_async_context(stdio_client(airbnb_server_params))
maps_client, _ = await stack.enter_async_context(stdio_client(maps_server_params))
airbnb_agent = Agent(
name="Airbnb",
role="Airbnb Agent",
model=OpenAIChat("gpt-4o"),
tools=[airbnb_client],
instructions=dedent("""\
你是一个可以在给定地点查找爱彼迎房源的代理。\
"""),
add_datetime_to_instructions=True,
)
为Agno AI代理集成MCP
6. 将MCP用于Upsonic代理
Upsonic是一个用于创建AI代理的Python框架。使用Upsonic,你可以构建自己的代理,为代理定义任务,并使用MCP工具处理每个任务定义,如下面的示例代码所示:
import os
from dotenv import load_dotenv
from upsonic import Task, Agent, Direct
from upsonic.client.tools import Search
load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
raise ValueError("OPENAI_API_KEY未在.env文件中找到")
os.environ["OPENAI_API_KEY"] = openai_api_key
class HackerNewsMCP:
command = "uvx"
args = ["mcp-hn"]
task = Task(
"分析今日HackerNews上排名前五的文章。简要总结每篇文章,"
"找出任何共同的主题或趋势,并突出哪些文章可能与对AI和软件开发感兴趣的人最相关。",
tools=[HackerNewsMCP, Search]
)
agent = Agent(
"科技新闻分析师",
company_url="https://news.ycombinator.com/",
company_objective="提供对科技行业新闻和趋势的深刻分析"
)
print("正在分析HackerNews文章...")
agent.print_do(task)
在上述示例中,我们在Upsonic中创建了一个AI代理,它从Hackernews中检索前五个最近的故事。
7. 为Mastra代理使用MCP
Mastra是一个TypeScript框架,用于构建原型和可投入生产的AI代理。与Chainlit类似,Mastra提供了一种标准化的方式来连接到MCP服务器,通过基于stdio或sse的连接访问广泛的工具。
要将Mastra代理连接到MCP服务器,你应该使用其 MCPConfiguration
类。此类在任何Mastra代理工作流中处理多个服务器连接,如生命周期、命名空间和工具管理。在Master应用程序和MCP服务器之间创建连接涉及以下步骤:
- 创建
MCPConfiguration
类的实例并添加服务器配置。 - 使用
getTools()
或getToolsets()
方法检索MCP工具。
以下示例代码展示了使用Mastra代理实现MCP服务器的基本用法:
import { MCPConfiguration } from "@mastra/mcp";
import { Agent } from "@mastra/core/agent";
import { openai } from "@ai-sdk/openai";
const mcp = new MCPConfiguration({
servers: {
stockPrice: {
command: "npx",
args: ["tsx", "stock-price.ts"],
env: {
API_KEY: "your-api-key",
},
},
weather: {
url: new URL("http://localhost:8080/sse"),
},
},
});
const agent = new Agent({
name: "多功能代理",
instructions: "你可以访问多个工具服务器。",
model: openai("gpt-4"),
tools: await mcp.getTools(),
});
LLM应用程序和代理中MCP的挑战和下一步是什么
本教程向你介绍了MCP,并解释了为什么它在开发者社区中如此受欢迎。我们强调了MCP与诸如Cursor和Windsurf等IDE的集成。此外,我们还在七种不同的Python和TypeScript框架中实现了MCP,用于构建基于LLM的应用程序、AI助手和代理。
MCP虽然强大,但也面临以下挑战。在为项目搜索MCP工具时,你可能会发现难以检查或验证工具的质量,也难以确定其是否适用于你的AI项目。这是因为其工具搜索和发现功能尚未标准化。此外,由于MCP服务器提供商的架构不同,其配置无法提供一致的用户体验。
目前,MCP生态系统正在讨论对MCP的各个方面进行标准化。未来可能会有一种标准的方法来安装基于MCP的应用程序,就像我们在Python中使用 pip install
安装软件包一样。PulseMCP也在努力使浏览和发现MCP工具变得更加容易。
评论