MCP Server를 npm에 올리고 다른 MCP와 함께 Amazon Bedrock에서 사용하기

지난 번 포스팅에서 사용했던 top_song을 npm에 올리고 다른 MCP Server와 함께 Bedrock에서 사용해보겠습니다.

간단한 MCP Server를 Amazon Bedrock에서 사용하기

우선 top_song은 js로 변경해서 npm에 올렸습니다.

@calc2te/top-song

npm에 올린 top_song을 테스트 하기 위해서 StdioServerParameters의 command와 args를 변경합니다.

server_params = StdioServerParameters(
    command="npx",
    args=['@calc2te/top-song'],
    env=None
)

실행하면 기존과 똑같이 잘 될 것입니다.

이제 기존에 만든 top-song과 다른 MCP Server까지 여러개를 사용해보도록 하겠습니다.

테스트를 위한 MCP Server는 위에서 만든 @calc2te/top-song과 @modelcontextprotocol/server-filesystem을 사용해보도록 하겠습니다.

MCP_SERVERS_CONFIG = {
    "top-song": {
        "command": "npx",
        "args": ["-y", "@calc2te/top-song"],
    },
    "filesystem": {
        "command": "npx",
        "args": [
            "-y",
            "@modelcontextprotocol/server-filesystem",
            "/Users/seobs/Documents/test"
        ]
    }
}

그리고 여러개를 사용할 수 있게 기존에 만든 mcp_client.py도 변경해보겠습니다.

기존에는 하나의 session만 필요했다면 이제는 여러개의 session이 필요하게 됩니다.
그래서 저는 clients에 각 session을 보관하려고 합니다.

async def connect_all(self):
    for server_name, config in self.server_configs.items():
        params = StdioServerParameters(
            command=config["command"],
            args=config.get("args", []),
            env=config.get("env", None)
        )
        
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(params))
        read_stream, write_stream = stdio_transport
        session = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))
        await session.initialize()
        
        self.clients[server_name] = session

사용 가능한 tool을 가져오는 부분도 조금 변경합니다.

async def list_all_tools(self):
    aggregated_tools = []
    for server_name, session in self.clients.items():
        tools_response = await session.list_tools()
        for tool in tools_response.tools:
            self.tool_mapping[tool.name] = server_name
            aggregated_tools.append({
                "toolSpec": {
                    "name": tool.name,
                    "description": tool.description,
                    "inputSchema": {
                        "json": tool.inputSchema
                    }
                }
            })
    return aggregated_tools

self.tool_mapping[tool.name] = server_name을 넣은 이유는 나중에 함수를 실행할때 서버의 이름이 필요합니다.

기존 코드에서는 session을 self.sesion으로 유지 했기 때문에 바로 사용이 가능했습니다.

result = await self.session.call_tool(tool_name, arguments=arguments)

 

그러나 지금처럼 MCP Server를 여러개 쓰면 self.session을 쓸 수가 없습니다.
게다가 Bedrock을 쓰게 되면 모델이 함수의 이름은 알려주지만 이게 어느 서버에 있는 함수인지는 알려주지 않습니다.

그래서 connect_all() 에서 각 session을 self.clients[server_name] = session을 통해 저장을 한 번 하고
list_all_tools() 에서는 self.tool_mapping[tool.name] = server_name을 통해 각 도구와 서버 이름을 매핑합니다.

그 이후에 알맞게 call_tool을 호출합니다.

async def call_tool(self, tool_name: str, arguments: dict):
    server_id = self.tool_mapping.get(tool_name)
    session = self.clients.get(server_id)
    result = await session.call_tool(tool_name, arguments=arguments)
    return result

이제 테스트를 해보겠습니다.

전체 코드는 Github의 client_for_multi_server 폴더를 확인해보세요.

GitHub - calc2te/mcp-and-amazon-bedrock