MCP Agent
Un servidor MCP local y ligero en Python que permite búsqueda RAG a través de AWS Lambda.
Documentación
Usando MCP Agent
MCP (Model Context Protocol) se está extendiendo rápidamente como una interfaz principal para que las aplicaciones de IA generativa utilicen datos externos. Aquí configuramos MCP y agentes en un entorno local para que puedas usar MCP fácilmente. Para más detalles como el despliegue en la nube, consulta kyopark/mcp.
Uso de MCP
MCP Básico
Los usuarios pueden conectarse a servidores MCP no solo a través de herramientas de IA instaladas en su computadora, como Claude Desktop o Cursor, sino también a través de aplicaciones desarrolladas principalmente en forma de agentes. El servidor MCP proporciona sus capacidades como capabilities en respuesta a las solicitudes del cliente MCP y ejecuta las solicitudes del cliente. El servidor MCP puede consultar archivos o bases de datos en la computadora local, así como consultar información necesaria utilizando APIs de servidores externos en Internet. El cliente MCP se conecta al servidor mediante el protocolo JSON-RPC 2.0, pudiendo elegir stdio o SSE (Server-Sent Events) para transmitir las solicitudes del host a MCP y recibir respuestas para utilizarlas.
La definición y el funcionamiento de los componentes principales de MCP son los siguientes:
- MCP Hosts: Programas/herramientas de IA que acceden a datos a través del protocolo MCP, como Claude Desktop, Cursor y User Agent Application.
- MCP Clients: Clientes que se conectan 1:1 con el servidor MCP, pudiendo conectarse mediante stdio o Streamable HTTP.
- MCP Servers: Programas ligeros que informan al cliente sobre las capacidades de las herramientas a través de MCP estandarizado, pudiendo consultar archivos o bases de datos en la computadora local y consultar información mediante APIs externas.
- Local data sources: Bases de datos y datos locales a los que el servidor MCP puede acceder.
- Remote services: Sistemas externos accesibles a través de APIs.
Usar MCP ofrece las siguientes ventajas:
- Acceso a diversas fuentes de datos de manera estandarizada.
- Se pueden agregar nuevas funciones mediante actualizaciones del servidor MCP sin cambiar el código de la aplicación.
- Facilita el soporte y la expansión de IA en toda la organización.
MCP Server Components incluye los siguientes elementos:
- Tools (Model-controlled): Funciones (herramientas) que el LLM puede invocar para realizar tareas específicas, ejecutando acciones concretas como una API.
tools = await session.list_tools()
- Resources (Application-controlled): Fuentes de datos a las que las aplicaciones de IA generativa pueden acceder. Pueden obtener datos sin cálculos significativos ni efectos secundarios.
resources = await session.list_resources()
- Prompts (User-controlled): Plantillas predefinidas que se utilizan al usar herramientas o recursos, y que se pueden seleccionar antes de la inferencia.
prompts = await session.list_prompts()
LangChain MCP Adapter
LangChain MCP Adapter es un envoltorio ligero que permite usar MCP con agentes LangGraph, y es un proyecto de código abierto bajo licencia MIT. El papel principal del adaptador MCP es definir herramientas para el servidor MCP, consultar la información de las herramientas desde el cliente MCP y ayudar a definirlas como nodos de herramientas de LangGraph para su uso.
MCP Server
El servidor MCP para búsqueda RAG se puede definir de la siguiente manera. Si se especifica el transporte del servidor como "stdio", el cliente puede ejecutar directamente el código Python del servidor sin necesidad de mantener el servidor en ejecución, lo cual es 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")
Cuando el servidor recibe una solicitud, realiza la búsqueda RAG con retrieve_knowledge_base(). Dado que el código Python del servidor debe ser ligero, se ha configurado para activar una lambda como se muestra a continuación. En Lambda se realizan las operaciones de retrieve, grade y generation. Se puede especificar "model_name" como se muestra, y opcionalmente usar "grading" según sea necesario. Además, si se desea acelerar mediante procesamiento paralelo, se establece "multi_region" en "Enable". Para el código detallado, consulta 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'], []
MCP Client
Si el cliente MCP solo ve un servidor MCP, se puede implementar usando stdio_client y StdioServerParameters como se muestra a continuación. La información del servidor MCP se puede leer desde config.json o usar la información ingresada por el usuario en streamlit. load_mcp_server_parameters() lee mcp_json y construye StdioServerParameters. La información del servidor MCP en config.json se obtiene de la salida generada después del despliegue con 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
)
A continuación, se configura stdio_client con la información del servidor MCP. En este punto, se obtiene la información de las herramientas con load_mcp_tools. En el agente, se vinculan las herramientas y se ejecuta la acción solicitada 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
El cliente MCP se ejecuta de la siguiente manera. Se utiliza asyncio para la ejecución asíncrona. Posteriormente, si el usuario actualiza la configuración de MCP en la interfaz de usuario, se puede actualizar la información.
asyncio.run(mcp_rag_agent_single(query, st))
Si hay varios servidores, se utiliza MultiServerMCPClient proporcionado por langchain-mcp-adapters. Primero, se obtiene la información del servidor como se muestra a continuación.
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
Luego, se define el cliente con la información del servidor MCP y MultiServerMCPClient como se muestra a continuación. La información de las herramientas obtenida del servidor MCP se recupera con client.get_tools() y se usa al crear el agente. Al igual que con un solo servidor MCP, se puede ejecutar con ainvoke para obtener 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
Aquí se ha definido el agente para facilitar la personalización.
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()
Uso de servidores MCP
Model Context Protocol servers también proporciona información sobre los siguientes servidores:
En Smithery puedes buscar servidores MCP y, si encuentras el que necesitas, consultar la información del servidor MCP en formato JSON para conectarte.
La información del servidor MCP para búsqueda de Google, consultada en Smithery - Google Search Server, es la siguiente. Requiere un ID de motor de búsqueda y una clave 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\"}"
]
}
}
}
Puedes actualizar la información del servidor en formato JSON como se muestra a continuación. Aquí se utiliza la búsqueda definida en mcp-server.py.
{
"mcpServers": {
"search": {
"command": "python",
"args": [
"application/mcp-server.py"
]
}
}
}
Ejecutar localmente (MAC)
-
Aunque no es obligatorio, se necesita AWS CLI para un desarrollo correcto. Instala siguiendo Instalar o actualizar la última versión de AWS CLI y luego registra las credenciales de AWS con el comando "aws configure".
-
Es conveniente configurar el entorno con venv. Crea una carpeta adecuada y configura el entorno como se muestra a continuación.
python -m venv venv
source venv/bin/activate
- Descarga el código fuente.
git clone https://github.com/kyopark2014/mcp-agent
- Luego, muévete a la carpeta de GitHub descargada e instala los paquetes necesarios como se muestra a continuación.
cd mcp-agent && python -m pip install -r requirements.txt
- Según Configuración de claves necesarias para el ejercicio, configura las claves para las APIs de internet y consulta del clima. Una vez configuradas, se generará un archivo JSON como el siguiente en application/config.json.
{
"WEATHER_API_KEY": "fbd00245cabcedefghijkd3e94905f7049",
"TAVILY_API_KEY": "tvly-1234567890U3imZFs4LNO2g0Qv1LoE"
}
- Ahora que todo está listo, ejecuta streamlit con el siguiente comando. Realiza pruebas de funcionamiento consultando Cómo usar las herramientas MCP.
streamlit run application/app.py
Ejecutar localmente con Docker
Instala y ejecuta Docker como se muestra a continuación.
brew install --cask docker
Ahora compila usando el script como se muestra a continuación. build.sh consulta las credenciales de AWS y las incluye al compilar.
./build.sh
Ahora ejecuta como se muestra a continuación. Aquí se ha configurado el puerto de Docker como 8502 por conveniencia, pero puedes ajustarlo según tu entorno.
docker run -p 8502:8501 mcp-agent
Accede a la siguiente URL en el navegador.
http://0.0.0.0:8502
Preparación para la ejecución
Para dibujar diagramas, instala graphviz siguiendo Graphviz. En Mac, usa el siguiente comando.
brew install graphviz
Resultados de la ejecución
Si realizas una pregunta compleja como "Quiero ir de Seúl a Jeju pasando por Busan. ¿Cuál es el clima y los restaurantes recomendados en el camino?", se recopila información utilizando varias herramientas como se muestra a continuación.
El resultado puede responder a preguntas complejas como se muestra a continuación.