昨天是用Gradio实现的MCP Server，只改一行代码让你的Gradio应用秒变MCP Server， 尝试用Gradio开发个Client。
代码
import asyncio
import gradio as gr
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP
from models import model
print(model.base_url)
# Define server options
server_options = [
"http://127.0.0.1:8000/sse",
"http://127.0.0.1:8000/mcp",
"http://127.0.0.1:7860/gradio_api/mcp/sse",
]
asyncdefprocess_query(server_url, query):
server = MCPServerHTTP(url=server_url)
agent = Agent(model, mcp_servers=[server])
asyncwith agent.run_mcp_servers():
result = await agent.run(query)
return result.output
defrun_async(server_url, query):
return asyncio.run(process_query(server_url, query))
# Create Gradio interface
with gr.Blocks(title="MCP Client") as demo:
gr.Markdown("# MCP Server Client")
with gr.Row():
server_dropdown = gr.Dropdown(
choices=server_options, value=server_options[0], label="MCP Server URL"
)
with gr.Row():
query_input = gr.Textbox(
lines=3, placeholder="Enter your query here...", label="Query"
)
with gr.Row():
submit_btn = gr.Button("Submit")
with gr.Row():
output_box = gr.Textbox(lines=5, label="Result")
submit_btn.click(
fn=run_async, inputs=[server_dropdown, query_input], outputs=output_box
)
if __name__ == "__main__":
demo.launch()
从上面代码看，用到了Gradio和PydanticAI。 而核心代码就是PydanticAI，去掉Gradio的代码部分，就是一个命令行的MCP Client
import asyncio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP
from models import model
asyncdefprocess_query(server_url, query):
server = MCPServerHTTP(url=server_url)
agent = Agent(model, mcp_servers=[server])
asyncwith agent.run_mcp_servers():
result = await agent.run(query)
return result.output
defrun_async(server_url, query):
return asyncio.run(process_query(server_url, query))
if __name__ == "__main__":
server_url="http://127.0.0.1:7860/gradio_api/mcp/sse"
query = "DeepSeek 有几个e，几个k？"
print(run_async(server_url,query))
接下来让我们重点介绍下PydanticAI。
PydanticAI
PydanticAI 是一个 Python 智能体框架，旨在减轻基于生成式 AI 构建生产级应用程序的开发负担。
为何使用 PydanticAI
• 由 Pydantic 团队打造：由 Pydantic[1]（OpenAI SDK、Anthropic SDK、LangChain、LlamaIndex、AutoGPT、Transformers、CrewAI、Instructor 等的验证层）背后的团队构建。 • 模型无关性：支持 OpenAI、Anthropic、Gemini、Deepseek、Ollama、Groq、Cohere 和 Mistral 等模型，并且提供简单接口以实现对 其他模型[2] 的支持。 • 与 Pydantic Logfire 集成：无缝 集成[3] 于 Pydantic Logfire[4] ，用于实时调试、性能监控以及跟踪你的 LLM 驱动应用程序的行为。 • 类型安全：旨在使 类型检查[5] 尽可能强大且富有信息性。 • 以 Python 为中心的设计：利用 Python 熟悉的控制流和代理组合来构建你的 AI 驱动项目，便于应用你在任何其他（非 AI）项目中使用的标准 Python 最佳实践。 • 结构化响应：借助 Pydantic[6] 的力量来 验证和结构化[7] 模型输出，确保响应在运行之间保持一致。 • 依赖注入系统：提供可选的 依赖注入[8] 系统，为你的代理的 系统提示[9]、工具[10] 和 输出验证器[11] 提供数据和服务。这对于测试和基于评估的迭代开发很有用。 • 流式响应：能够 流式传输[12] LLM 响应，进行即时验证，确保实时获取经过验证的输出。 • 图支持：Pydantic Graph[13] 提供一种使用类型提示定义图的强大方式，在标准控制流可能退化为意大利面条代码的复杂应用中非常有用。
hello world
from pydantic_ai import Agent
agent = Agent(
'google-gla:gemini-1.5-flash',
system_prompt='Be concise, reply with one sentence.',
)
result = agent.run_sync('Where the olympics held in 2008?')
print(result.output)
"""
The 2008 Olympics were held in Beijing, China.
"""
例子是使用Google的gemini-1.5-flash， 如果我想用ollama的模型怎么做？
1. 选择有tool标签的模型[14] 比如最新的qwen3， 如果用deepseek-r1:32b， 会报如下错误，
openai.BadRequestError: Error code: 400 - {'error': {'message': 'registry.ollama.ai/library/deepseek-r1:32b does not support tools', 'type': 'api_error', 'param': None, 'code': None}}
2. 更新 hello_world.py
代码
from pydantic_ai import Agent
from models import model
agent = Agent(
model,
system_prompt='Be concise, reply with one sentence.',
)
result = agent.run_sync('Where the olympics held in 2008?')
print(result.output)
"""
The 2008 Olympics were held in Beijing, China.
"""
将 'google-gla:gemini-1.5-flash'
替换为model
期中model的实现如下，
• models.py
import os
from dotenv import load_dotenv
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
load_dotenv()
base_url, api_key, model_name = (
os.getenv("BASE_URL"),
os.getenv("API_KEY"),
os.getenv("MODEL_NAME"),
)
provider = OpenAIProvider(base_url=base_url, api_key=api_key)
model = OpenAIModel(model_name, provider=provider)
而 .env
内容如下,
BASE_URL=http://127.0.0.1:11434/v1
API_KEY=ollama
MODEL_NAME=qwen3:32b
执行
python hello_world.py
会返回
The 2008 Olympics were held in Beijing, China.
3. 如果电脑性能不够，或者不想自己安装 OLLAMA 等部署大模型的应用。推荐注册硅基流动（ SiliconCloud）
https://cloud.siliconflow.cn/i/kfX7YFRm SiliconCloud 邀请奖励持续进行，2000 万 Tokens 送不停！ 1. 邀请好友赚 2000 万 Tokens：每成功邀请一位新用户通过手机号码注册，您将获得 2000 万 Tokens； 2. 注册即送 2000 万 Tokens：受邀好友作为新用户完成 SiliconCloud 账号注册，立刻获得 2000万 Tokens。
则.env
的内容类似如下，
API_KEY 是你自己创建的API密钥 模型记得选择支持tool的，下图是选择了支持tool，并且免费的模型。BASE_URL=https://api.closeai.im/v1
API_KEY=sk-00Q297d0i3i5p7ZY49Ll54VqHqf2Q1SlJgdpaXHV4R9s2siK
MODEL_NAME=Qwen/Qwen3-32B
hello_world.py
增加格式化输出
from typing importUnion
from pydantic import BaseModel
from pydantic_ai import Agent
from models import model
classCityLocation(BaseModel):
city: str
country: str
agent = Agent(
model,
system_prompt="Be concise, reply with one sentence.",
output_type=Union[CityLocation, str],
)
result = agent.run_sync("Where the olympics held in 2008?")
print(result.output)
再次运行python hello_world.py
返回
city='Beijing' country='China'
使用Gradio聊天机器人作为MCP客户端
Gradio 官网提供了一个使用Gradio聊天机器人作为MCP客户端[15]的教程，有兴趣可以看看实现。
我这里提供一个简化版本的chat_bot.py
import asyncio
import gradio as gr
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP
from models import model
asyncdefprocess_query(server_url, query):
server = MCPServerHTTP(url=server_url)
agent = Agent(model, mcp_servers=[server], system_prompt="/no_think")
asyncwith agent.run_mcp_servers():
result = await agent.run(query)
return result.output
server_url = "http://127.0.0.1:7860/gradio_api/mcp/sse"
defrun_async(query, history):
return asyncio.run(process_query(server_url, query))
gr.ChatInterface(fn=run_async, type="messages").launch()
这个比开始的实现更简单。
基于gradio chatbot的的MCP client
引用链接
[1]
Pydantic:https://docs.pydantic.dev/latest/[2]
其他模型:https://ai.pydantic.dev//models/[3]
集成:https://ai.pydantic.dev//logfire/[4]
Pydantic Logfire:https://pydantic.dev/logfire[5]
类型检查:https://ai.pydantic.dev//agents/#static-type-checking[6]
Pydantic:https://docs.pydantic.dev/latest/[7]
验证和结构化:https://ai.pydantic.dev//output/#structured-output[8]
依赖注入:https://ai.pydantic.dev//dependencies/[9]
系统提示:https://ai.pydantic.dev//agents/#system-prompts[10]
工具:https://ai.pydantic.dev//tools/[11]
输出验证器:https://ai.pydantic.dev//output/#output-validator-functions[12]
流式传输:https://ai.pydantic.dev//output/#streamed-results[13]
Pydantic Graph:https://ai.pydantic.dev//graph/[14]
有tool标签的模型:https://ollama.com/search?c=tools[15]
使用Gradio聊天机器人作为MCP客户端:https://www.gradio.app/guides/building-an-mcp-client-with-gradio
