Use ferramentas MCP em um agente Python
Conecte-se a um servidor MCP, descubra suas ferramentas e execute um agente Python que as utilize. Use a URL de um MCP fornecido pela Databricks, do seu MCP registrado ou do seu servidor no Databricks Apps.
Para usar MCPs do Claude Code, Codex ou outro agente de codificação, escolha seu cliente em Agentes de codificação compatíveis. Para outros assistentes e clientes MCP, consulte Outros clientes MCP.
Para um projeto de CLI do Agent Bricks, adicione ferramentas MCP com a CLI. Os exemplos abaixo mostram como se conectar a partir do seu próprio código em Python.
Pré-requisitos
- Python 3.12 em seu computador.
- A URL do MCP do seu servidor. Se alguém compartilhou um MCP com você, essa pessoa deve conceder acesso a você. Conclua o login do provedor se ele usar OAuth por usuário. Para um servidor no Databricks Apps, você precisa de CAN USE no aplicativo.
- Acesso a um endpoint de modelo que ofereça suporte à chamada de ferramentas do Databricks. O exemplo usa
databricks-claude-sonnet-4-5. Substitua-o por um endpoint disponível em seu workspace.
Para usar system.ai.dbsql, system.ai.sandbox ou system.ai.web_search, um administrador de account deve habilitar a versão beta do Unity Gateway na página Previews do console da account. Consulte Gerenciar prévias da account.
Para opções de identidade e requisitos de conectividade, consulte Authentication and network access.
o passo 1: Instalar e entrar
-
Se você ainda não instalou a CLI do Databricks, execute este comando no macOS ou Linux:
Bashcurl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | shPara o Windows ou outros métodos de instalação, consulte Instalar a CLI do Databricks.
-
Faça login no seu Workspace:
Bashdatabricks auth login --host https://<workspace-hostname> --profile DEFAULT -
Instale as bibliotecas do Python:
Bashpip install --upgrade databricks-mcp databricks-sdk "mcp>=1.24,<2"
Os exemplos usam o MCP Python 1.x, que é compatível com os frameworks de agentes abaixo.
O passo 2: Conectar ao seu servidor
Salve o seguinte código como mcp_agent.py. Substitua <mcp-server-url> pela URL do seu servidor:
- MCP fornecido ou registrado pela Databricks:
https://<workspace-hostname>/ai-gateway/mcp-services/<catalog>.<schema>.<service>. - Servidor no Databricks Apps: copie o URL do aplicativo em sua página de visão geral e adicione
/mcp.
from databricks_mcp import DatabricksMCPClient
from databricks.sdk import WorkspaceClient
workspace_client = WorkspaceClient()
server_url = "<mcp-server-url>"
mcp_client = DatabricksMCPClient(
server_url=server_url,
workspace_client=workspace_client,
)
tools = mcp_client.list_tools()
for tool in tools:
print(tool.name, tool.description, tool.inputSchema, sep="\n")
Faça a execução do script:
python mcp_agent.py
Escolha uma tarefa somente de leitura compatível com uma dessas ferramentas para o próximo passo. Escolha uma tarefa somente leitura compatível com uma dessas ferramentas para o próximo o passo. Se a lista estiver vazia ou a conexão falhar, consulte Autenticação e rede do MCP.
Use outro servidor
- Para um MCP integrada, use seu nome completo no URL do MCP, como
system.ai.github. - Para uma integração existente com um servidor MCP de workspace legado, substitua
server_urlpela URL do endpoint legado correspondente.
Para encontrar MCPs que você pode acessar em um catálogo e esquema, execução:
databricks ai-gateway list-mcp-services --parent schemas/system.ai
Substitua system.ai pelo seu <catalog>.<schema> para listar os MCPs registrados. A CLI lida com a paginação.
O passo 3: Execução um agente com essas ferramentas
Escolha sua estrutura, instale o pacote e adicione seu exemplo em Python a mcp_agent.py. Cada exemplo usa o mesmo server_url e o logon do workspace do o passo 2.
- LangGraph
- OpenAI Agents SDK
- OpenAI client
O exemplo converte blocos de conteúdo MCP para o formato de mensagem de chat do modelo com convert_to_openai_messages.
pip install --upgrade databricks-langchain langgraph
import asyncio
from databricks_langchain import (
ChatDatabricks,
DatabricksMCPServer,
DatabricksMultiServerMCPClient,
)
from langchain_core.messages import convert_to_openai_messages
from langgraph.prebuilt import create_react_agent
async def main():
client = DatabricksMultiServerMCPClient([
DatabricksMCPServer(
name="my-mcp-server",
url=server_url,
workspace_client=workspace_client,
),
])
agent = create_react_agent(
ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
tools=await client.get_tools(),
prompt=lambda state: convert_to_openai_messages(state["messages"]),
)
task = input("Ask the agent to use a tool: ")
result = await agent.ainvoke({
"messages": [{"role": "user", "content": task}],
})
for message in result["messages"]:
print(message)
asyncio.run(main())
Notebook de implantação (opcional)
Para implantação de Model Serving, adapte este notebook para usar a URL do seu servidor:
LangGraph MCP tool-calling agent
pip install --upgrade databricks-openai openai-agents
import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from agents.mcp import MCPServerStreamableHttpParams
from agents.tracing import set_trace_processors
from databricks_openai import AsyncDatabricksOpenAI
from databricks_openai.agents.mcp_server import McpServer
set_default_openai_client(AsyncDatabricksOpenAI())
set_default_openai_api("chat_completions")
set_trace_processors([])
async def main():
async with McpServer(
name="my-mcp-server",
params=MCPServerStreamableHttpParams(url=server_url),
workspace_client=workspace_client,
) as server:
agent = Agent(
name="Tool-using agent",
instructions="Use the available tools to answer the user's question.",
model="databricks-claude-sonnet-4-5",
mcp_servers=[server],
)
task = input("Ask the agent to use a tool: ")
result = await Runner.run(agent, task)
for item in result.new_items:
print(item.to_input_item())
asyncio.run(main())
Notebook de implantação (opcional)
Para implantação de Model Serving, adapte este notebook para usar a URL do seu servidor:
Agents SDK MCP tool-calling agent
pip install --upgrade databricks-openai
Este exemplo execução o loop de chamada de ferramenta explicitamente usando o cliente compatível com OpenAI.
import json
from databricks_openai import DatabricksOpenAI, McpServerToolkit
toolkit = McpServerToolkit(url=server_url, workspace_client=workspace_client)
tools_by_name = {tool.name: tool for tool in toolkit.get_tools()}
model_client = DatabricksOpenAI()
messages = [{"role": "user", "content": input("Ask the agent to use a tool: ")}]
for _ in range(10):
response = model_client.chat.completions.create(
model="databricks-claude-sonnet-4-5",
messages=messages,
tools=[tool.spec for tool in tools_by_name.values()],
)
message = response.choices[0].message
messages.append(message.model_dump(exclude_none=True))
if not message.tool_calls:
print(message.content)
break
for call in message.tool_calls:
try:
tool = tools_by_name.get(call.function.name)
if tool is None:
raise ValueError(f"Unknown tool: {call.function.name}")
arguments = json.loads(call.function.arguments or "{}")
if not isinstance(arguments, dict):
raise ValueError("Tool arguments must be a JSON object.")
output = tool.execute(**arguments)
except Exception as error:
output = json.dumps({"error": str(error)})
print(call.function.name, output)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": str(output),
})
else:
raise RuntimeError("The agent reached the tool-calling limit.")
Notebook de implantação (opcional)
Para implantação de Model Serving, adapte este notebook para usar a URL do seu servidor:
OpenAI MCP tool-calling agent
Execute python mcp_agent.py novamente. Quando solicitado, peça a tarefa somente leitura que você escolheu, incluindo quaisquer entradas necessárias. Por exemplo, se o seu servidor tiver uma ferramenta de busca de tickets, peça a ela para localizar tickets abertos em um projeto específico.
Verifique a conversa impressa para ver uma chamada de ferramenta, seu resultado e a resposta do agente. Uma resposta sem uma chamada de ferramenta não confirma que o servidor MCP foi usado.
Chamar uma ferramenta diretamente para solucionar problemas
Execute-o após o código de conexão no o passo 2: Faça a execução após o código de conexão do passo 2:
import json
tool_name = input("Read-only tool name: ")
arguments = json.loads(input("Tool arguments as a JSON object: "))
result = mcp_client.call_tool(tool_name, arguments)
print(result)
Verifique se o resultado não apresenta erro de ferramenta e contém os dados esperados.
Descubra nomes de ferramentas e esquemas de entrada com list_tools() antes de chamar uma ferramenta. Os formatos de resultado variam de acordo com a ferramenta:
- Se
structuredContentestiver presente, use esse resultado estruturado diretamente. Uma ferramenta pode descrever sua forma comoutputSchema. - Caso contrário, inspecione os blocos
content. Analise um bloco de texto como JSON somente se a ferramenta retornar JSON. O MCP também oferece suporte a texto simples e outros tipos de conteúdo. - Verifique
isErrore inspecione uma resposta de amostra antes de depender de campos de saída específicos.
Implante e compartilhe quando estiver tudo pronto
O exemplo local execução como você. Quando você implantar o agente no Databricks Apps, escolha a identidade que ele usa: o Service Principal do aplicativo para acesso compartilhado ou o usuário solicitante para acesso por usuário.
Para MCPs fornecidos ou registrados pela Databricks:
- Conceda ao chamador acesso ao MCP e ao catálogo e esquema pai dele.
- Configure o acesso por usuário se o seu agente agir em nome de um usuário.
- Govern the MCP to restrict tools, apply policies, set rate limits, and monitor calls.
Para servidores de workspace legados ou servidores hospedados em Databricks Apps, conceda acesso aos recursos subjacentes ou ao aplicativo. Consulte Autenticação de agente.