MCP Agent
Um servidor MCP leve e local em Python que permite pesquisa RAG através do AWS Lambda.
Documentação
Utilizando o MCP Agent
O MCP (Model Context Protocol) está se espalhando rapidamente como uma interface principal para aplicações de IA generativa utilizarem dados externos. Aqui, configuramos o MCP e o agente em um ambiente local para facilitar o uso do MCP. Para mais detalhes, como implantação em nuvem, consulte kyopark/mcp.
Utilização do MCP
MCP Básico
Os usuários podem se conectar a servidores MCP não apenas por meio de ferramentas de IA instaladas em seus computadores, como Claude Desktop e Cursor, mas também principalmente por meio de aplicativos desenvolvidos na forma de agentes. O servidor MCP fornece suas capacidades como capabilities em resposta às solicitações do cliente MCP e executa as solicitações do cliente. O servidor MCP pode consultar arquivos ou bancos de dados do computador local, bem como consultar informações necessárias usando APIs de servidores externos na internet. O Cliente MCP se conecta ao Servidor usando o protocolo JSON-RPC 2.0, podendo escolher entre stdio ou SSE (Server-Sent Events) para transmitir as solicitações do Host ao MCP e receber e utilizar as respostas.
A definição e o funcionamento dos principais elementos do MCP são os seguintes:
- MCP Hosts: Programas/ferramentas de IA que acessam dados por meio do protocolo MCP, incluindo Claude Desktop, Cursor e User Agent Application.
- MCP Clients: Clientes que se conectam 1:1 com o Servidor MCP, podendo se conectar via stdio ou Streamable HTTP.
- MCP Servers: Programas leves que informam as capacidades das ferramentas ao Cliente por meio do MCP padronizado, podendo consultar arquivos ou bancos de dados do computador local e também consultar informações usando APIs externas.
- Local data sources: Bancos de dados e dados locais que o servidor MCP pode acessar
- Remote services: Sistemas externos acessíveis via API
Usar o MCP oferece as seguintes vantagens:
- Acesso a diversas fontes de dados de forma padronizada.
- Adição de novos recursos por meio de atualizações do servidor MCP sem alterar o código do aplicativo.
- Facilidade de suporte e expansão de IA em toda a organização.
MCP Server Components inclui os seguintes itens:
- Tools (Model-controlled): Funções (ferramentas) que o LLM pode chamar para executar tarefas específicas, realizando ações como uma API.
tools = await session.list_tools()
- Resources (Application-controlled): Fontes de dados que aplicações de IA generativa podem acessar. Podem obter dados sem cálculos significativos ou efeitos colaterais.
resources = await session.list_resources()
- Prompts (User-controlled): Modelos predefinidos usados ao utilizar ferramentas ou recursos, que podem ser selecionados antes da inferência.
prompts = await session.list_prompts()
Adaptador MCP LangChain
LangChain MCP Adapter é um wrapper leve que permite usar o MCP com agentes LangGraph, sendo um projeto de código aberto baseado em MIT. O principal papel do Adaptador MCP é definir ferramentas para servidores MCP, consultar informações das ferramentas no cliente MCP e ajudar a defini-las como nós de ferramentas no LangGraph para uso.
Servidor MCP
O servidor MCP para busca RAG pode ser definido da seguinte forma. Se o transporte do servidor for definido como "stdio", o cliente pode executar diretamente o código Python do servidor sem precisar mantê-lo em execução, o que é conveniente.
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name = "Search",
instructions=(
"You are a helpful assistant. "
"You can search the documentation for the user's question and provide the answer."
),
)
@mcp.tool()
def search(keyword: str) -> str:
"search keyword"
return retrieve_knowledge_base(keyword)
if __name__ =="__main__":
print(f"###### main ######")
mcp.run(transport="stdio")
Quando o servidor recebe uma solicitação, ele executa a busca RAG com retrieve_knowledge_base(). Como o código Python do servidor deve ser leve, ele foi configurado para acionar uma lambda, como mostrado abaixo. A lambda executa as operações de retrieve, grade e generation. Você pode especificar "model_name" como abaixo e, se necessário, usar "grading" opcionalmente. Além disso, se desejar acelerar o processamento com paralelismo, defina "multi_region" como "Enable". Para o código detalhado, consulte lambda-rag.
def retrieve_knowledge_base(query):
lambda_client = boto3.client(
service_name='lambda',
region_name=bedrock_region
)
functionName = f"lambda-rag-for-{projectName}"
payload = {
'function': 'search_rag',
'knowledge_base_name': knowledge_base_name,
'keyword': query,
'top_k': numberOfDocs,
'grading': "Enable",
'model_name': model_name,
'multi_region': multi_region
}
output = lambda_client.invoke(
FunctionName=functionName,
Payload=json.dumps(payload),
)
payload = json.load(output['Payload'])
return payload['response'], []
Cliente MCP
Se o cliente MCP visualizar apenas um servidor MCP, ele pode ser implementado usando stdio_client e StdioServerParameters, como abaixo. As informações sobre o servidor MCP podem ser lidas do config.json ou usar informações inseridas pelo usuário no streamlit. load_mcp_server_parameters() lê o mcp_json e constrói StdioServerParameters. As informações do servidor MCP no config.json são obtidas da saída gerada após a implantação com AWS CDK.
from mcp import ClientSession, StdioServerParameters
def load_mcp_server_parameters():
mcp_json = json.loads(mcp_config)
mcpServers = mcp_json.get("mcpServers")
command = ""
args = []
if mcpServers is not None:
for server in mcpServers:
config = mcpServers.get(server)
if "command" in config:
command = config["command"]
if "args" in config:
args = config["args"]
break
return StdioServerParameters(
command=command,
args=args
)
Configure o stdio_client com as informações do servidor MCP, como abaixo. Nesse caso, as informações sobre as ferramentas são obtidas com load_mcp_tools. No agente, as informações da ferramenta são vinculadas e a ação solicitada é executada usando ainvoke.
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
async def mcp_rag_agent_single(query, st):
server_params = load_mcp_server_parameters()
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await load_mcp_tools(session)
with st.status("thinking...", expanded=True, state="running") as status:
agent = create_agent(tools)
agent_response = await agent.ainvoke({"messages": query})
result = agent_response["messages"][-1].content
st.markdown(result)
st.session_state.messages.append({
"role": "assistant",
"content": result
})
return result
O cliente MCP é executado da seguinte forma. Usamos asyncio para execução assíncrona. Depois, se o usuário atualizar a Configuração MCP na interface, as informações podem ser atualizadas.
asyncio.run(mcp_rag_agent_single(query, st))
Quando há várias informações de servidor, use o MultiServerMCPClient fornecido por langchain-mcp-adapters. Primeiro, obtenha as informações do servidor como abaixo.
def load_multiple_mcp_server_parameters():
mcp_json = json.loads(mcp_config)
mcpServers = mcp_json.get("mcpServers")
server_info = {}
if mcpServers is not None:
command = ""
args = []
for server in mcpServers:
config = mcpServers.get(server)
if "command" in config:
command = config["command"]
if "args" in config:
args = config["args"]
server_info[server] = {
"command": command,
"args": args,
"transport": "stdio"
}
return server_info
Em seguida, defina o cliente com as informações do servidor MCP e o MultiServerMCPClient, como abaixo. As informações das ferramentas obtidas do servidor MCP são recuperadas com client.get_tools() e usadas ao criar o agente. Assim como no servidor MCP único, você pode executar com ainvoke e obter resultados.
from langchain_mcp_adapters.client import MultiServerMCPClient
asyncio.run(mcp_rag_agent_multiple(query, st))
async def mcp_rag_agent_multiple(query, st):
server_params = load_multiple_mcp_server_parameters()
async with MultiServerMCPClient(server_params) as client:
with st.status("thinking...", expanded=True, state="running") as status:
tools = client.get_tools()
agent = create_agent(tools)
response = await agent.ainvoke({"messages": query})
result = response["messages"][-1].content
st.markdown(result)
st.session_state.messages.append({
"role": "assistant",
"content": result
})
return result
Aqui, definimos o agente para facilitar a personalização.
def create_agent(tools):
tool_node = ToolNode(tools)
chatModel = get_chat(extended_thinking="Disable")
model = chatModel.bind_tools(tools)
class State(TypedDict):
messages: Annotated[list, add_messages]
def call_model(state: State, config):
system = (
"당신의 이름은 서연이고, 질문에 친근한 방식으로 대답하도록 설계된 대화형 AI입니다."
"상황에 맞는 구체적인 세부 정보를 충분히 제공합니다."
"모르는 질문을 받으면 솔직히 모른다고 말합니다."
"한국어로 답변하세요."
)
try:
prompt = ChatPromptTemplate.from_messages(
[
("system", system),
MessagesPlaceholder(variable_name="messages"),
]
)
chain = prompt | model
response = chain.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: State) -> Literal["continue", "end"]:
messages = state["messages"]
last_message = messages[-1]
if isinstance(last_message, AIMessage) and last_message.tool_calls:
return "continue"
else:
return "end"
def buildChatAgent():
workflow = StateGraph(State)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"end": END,
},
)
workflow.add_edge("action", "agent")
return workflow.compile()
return buildChatAgent()
Utilização de Servidores MCP
Model Context Protocol servers também fornece informações sobre os seguintes servidores:
No Smithery, você pode procurar servidores MCP e, ao encontrar o servidor necessário, consultar as informações do servidor MCP em formato JSON.
As informações do servidor MCP para pesquisa do Google, verificadas em Smithery - Google Search Server, são as seguintes. Elas exigem o ID do mecanismo de pesquisa e a chave da API.
{
"mcpServers": {
"google-search-mcp-server": {
"command": "npx",
"args": [
"-y",
"@smithery/cli@latest",
"run",
"@gradusnikov/google-search-mcp-server",
"--config",
"{\"googleCseId\":\"b5cd8c527fbd64b72\",\"googleApiKey\":\"AIzbSyDQlYpck8-9TbBSuxoew1luOGVB6unRPNk\"}"
]
}
}
}
Você pode atualizar as informações do servidor em formato JSON, como abaixo. Aqui, usamos a função search definida em mcp-server.py.
{
"mcpServers": {
"search": {
"command": "python",
"args": [
"application/mcp-server.py"
]
}
}
}
Executando localmente (MAC)
-
Embora não seja obrigatório, o AWS CLI é necessário para um andamento adequado. Instale seguindo Instalar ou atualizar a versão mais recente do AWS CLI e registre as credenciais AWS com o comando "aws configure".
-
Configurar o ambiente com venv é conveniente. Crie uma pasta adequada e configure o ambiente como abaixo.
python -m venv venv
source venv/bin/activate
- Baixe o código-fonte.
git clone https://github.com/kyopark2014/mcp-agent
- Em seguida, navegue até a pasta do GitHub baixada e instale os pacotes necessários, como abaixo.
cd mcp-agent && python -m pip install -r requirements.txt
- De acordo com Configuração de chaves necessárias para o laboratório, defina as chaves para a API de internet e consulta de clima. Após a configuração, o arquivo application/config.json será gerado como o seguinte JSON.
{
"WEATHER_API_KEY": "fbd00245cabcedefghijkd3e94905f7049",
"TAVILY_API_KEY": "tvly-1234567890U3imZFs4LNO2g0Qv1LoE"
}
- Agora que está pronto, execute o streamlit com o comando abaixo. Consulte Como usar as ferramentas MCP para testar o funcionamento.
streamlit run application/app.py
Executando localmente com Docker
Instale e execute o Docker como abaixo.
brew install --cask docker
Agora, faça o build usando o script abaixo. build.sh consulta as credenciais AWS e as inclui no build.
./build.sh
Agora, execute como abaixo. Aqui, definimos a porta do Docker como 8502 por conveniência, mas você pode configurá-la de acordo com seu ambiente.
docker run -p 8502:8501 mcp-agent
Acesse a URL abaixo no navegador.
http://0.0.0.0:8502
Preparação para execução
Para desenhar diagramas, instale o Graphviz seguindo Graphviz. No Mac, use o comando abaixo.
brew install graphviz
Resultados da execução
Se você fizer uma pergunta complexa como "Quero ir de Seul a Jeju passando por Busan. Qual é o clima e os restaurantes durante o caminho?", as informações serão coletadas usando várias ferramentas, como abaixo.
O resultado pode responder a perguntas complexas, como abaixo.