Skip to content

manager

Class info

Classes

Name Children Inherits
MCPClient
llmling_agent.mcp_server.client
MCP client for communicating with MCP servers.
    MCPManager
    llmling_agent.mcp_server.manager
    Manages MCP server connections and tools.
      ResourceInfo
      llmling_agent_config.resources
      Information about an available resource.
        ResourceProvider
        llmling_agent.resource_providers.base
        Base class for resource providers.
        SSEMCPServerConfig
        llmling_agent_config.mcp_server
        MCP server using Server-Sent Events transport.
          • BaseMCPServerConfig
          StdioMCPServerConfig
          llmling_agent_config.mcp_server
          MCP server started via stdio.
            • BaseMCPServerConfig
            StreamableHTTPMCPServerConfig
            llmling_agent_config.mcp_server
            MCP server using StreamableHttp.
              • BaseMCPServerConfig

              🛈 DocStrings

              Tool management for LLMling agents.

              MCPManager

              Bases: ResourceProvider

              Manages MCP server connections and tools.

              Source code in src/llmling_agent/mcp_server/manager.py 
               56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
              class MCPManager(ResourceProvider):
    """Manages MCP server connections and tools."""

    def __init__(
        self,
        name: str = "mcp",
        owner: str | None = None,
        servers: Sequence[MCPServerConfig | str] | None = None,
        context: NodeContext | None = None,
    ):
        super().__init__(name, owner=owner)
        self.servers: list[MCPServerConfig] = []
        for server in servers or []:
            self.add_server_config(server)
        self.context = context
        self.clients: dict[str, MCPClient] = {}
        self.exit_stack = AsyncExitStack()

    @property
    def requires_async(self) -> bool:
        return True

    def add_server_config(self, server: MCPServerConfig | str):
        """Add a new MCP server to the manager."""
        server = (
            StdioMCPServerConfig.from_string(server)
            if isinstance(server, str)
            else server
        )
        self.servers.append(server)

    def __repr__(self) -> str:
        return f"MCPManager({self.servers!r})"

    async def __aenter__(self) -> Self:
        try:
            # Setup directly provided servers
            for server in self.servers:
                await self.setup_server(server)

            # Setup servers from context if available
            if self.context and self.context.config and self.context.config.mcp_servers:
                for server in self.context.config.get_mcp_servers():
                    await self.setup_server(server)

        except Exception as e:
            # Clean up in case of error
            await self.__aexit__(type(e), e, e.__traceback__)
            msg = "Failed to initialize MCP manager"
            raise RuntimeError(msg) from e

        return self

    async def __aexit__(self, *exc):
        await self.cleanup()

    async def setup_server(self, config: MCPServerConfig):
        """Set up a single MCP server connection."""
        if not config.enabled:
            return
        env = config.get_env_vars()
        match config:
            case StdioMCPServerConfig():
                client = MCPClient(transport_mode="stdio")
                client = await self.exit_stack.enter_async_context(client)
                await client.connect(config.command, args=config.args, env=env)
                client_id = f"{config.command}_{' '.join(config.args)}"
            case SSEMCPServerConfig():
                client = MCPClient(transport_mode="sse")
                client = await self.exit_stack.enter_async_context(client)
                await client.connect("", [], url=config.url, env=env)
                client_id = f"sse_{config.url}"
            case StreamableHTTPMCPServerConfig():
                client = MCPClient(transport_mode="streamable-http")
                client = await self.exit_stack.enter_async_context(client)
                await client.connect("", [], url=config.url, env=env)
                client_id = f"streamable_http_{config.url}"
        self.clients[client_id] = client

    async def get_tools(self) -> list[Tool]:
        """Get all tools from all connected servers."""
        from llmling_agent.tools.base import Tool

        tools: list[Tool] = []
        for client in self.clients.values():
            for tool in client._available_tools:
                try:
                    fn = client.create_tool_callable(tool)
                    meta = {"mcp_tool": tool.name}
                    tool_info = Tool.from_callable(fn, source="mcp", metadata=meta)
                    tools.append(tool_info)
                    logger.debug("Registered MCP tool: %s", tool.name)
                except Exception:
                    msg = "Failed to create tool from MCP tool: %s"
                    logger.exception(msg, tool.name)
                    continue

        return tools

    async def list_prompts(self) -> list[StaticPrompt]:
        """Get all available prompts from MCP servers."""
        prompts = []
        for client in self.clients.values():
            try:
                result = await client.list_prompts()
            except Exception:
                logger.exception("Failed to get prompts from MCP server")
            for prompt in result.prompts:
                try:
                    converted = await convert_mcp_prompt(client, prompt)
                    prompts.append(converted)
                except Exception:
                    logger.exception("Failed to convert prompt: %s", prompt.name)
        return prompts

    async def list_resources(self) -> list[ResourceInfo]:
        """Get all available resources from MCP servers."""
        resources = []
        for client in self.clients.values():
            try:
                result = await client.list_resources()
            except Exception:
                logger.exception("Failed to get resources from MCP server")
            for resource in result.resources:
                try:
                    converted = await convert_mcp_resource(resource)
                    resources.append(converted)
                except Exception:
                    logger.exception("Failed to convert resource: %s", resource.name)
        return resources

    async def cleanup(self):
        """Clean up all MCP connections."""
        try:
            try:
                # Clean up exit stack (which includes MCP clients)
                await self.exit_stack.aclose()
            except RuntimeError as e:
                if "different task" in str(e):
                    # Handle task context mismatch
                    current_task = asyncio.current_task()
                    if current_task:
                        loop = asyncio.get_running_loop()
                        await loop.create_task(self.exit_stack.aclose())
                else:
                    raise

            self.clients.clear()

        except Exception as e:
            msg = "Error during MCP manager cleanup"
            logger.exception(msg, exc_info=e)
            raise RuntimeError(msg) from e

    @property
    def active_servers(self) -> list[str]:
        """Get IDs of active servers."""
        return list(self.clients)

              active_servers property 

              active_servers: list[str]

              Get IDs of active servers.

              add_server_config 

              add_server_config(server: MCPServerConfig | str)

              Add a new MCP server to the manager.

              Source code in src/llmling_agent/mcp_server/manager.py 
              78
79
80
81
82
83
84
85
              def add_server_config(self, server: MCPServerConfig | str):
    """Add a new MCP server to the manager."""
    server = (
        StdioMCPServerConfig.from_string(server)
        if isinstance(server, str)
        else server
    )
    self.servers.append(server)

              cleanup async 

              cleanup()

              Clean up all MCP connections.

              Source code in src/llmling_agent/mcp_server/manager.py 
              187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
              async def cleanup(self):
    """Clean up all MCP connections."""
    try:
        try:
            # Clean up exit stack (which includes MCP clients)
            await self.exit_stack.aclose()
        except RuntimeError as e:
            if "different task" in str(e):
                # Handle task context mismatch
                current_task = asyncio.current_task()
                if current_task:
                    loop = asyncio.get_running_loop()
                    await loop.create_task(self.exit_stack.aclose())
            else:
                raise

        self.clients.clear()

    except Exception as e:
        msg = "Error during MCP manager cleanup"
        logger.exception(msg, exc_info=e)
        raise RuntimeError(msg) from e

              get_tools async 

              get_tools() -> list[Tool]

              Get all tools from all connected servers.

              Source code in src/llmling_agent/mcp_server/manager.py 
              135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
              async def get_tools(self) -> list[Tool]:
    """Get all tools from all connected servers."""
    from llmling_agent.tools.base import Tool

    tools: list[Tool] = []
    for client in self.clients.values():
        for tool in client._available_tools:
            try:
                fn = client.create_tool_callable(tool)
                meta = {"mcp_tool": tool.name}
                tool_info = Tool.from_callable(fn, source="mcp", metadata=meta)
                tools.append(tool_info)
                logger.debug("Registered MCP tool: %s", tool.name)
            except Exception:
                msg = "Failed to create tool from MCP tool: %s"
                logger.exception(msg, tool.name)
                continue

    return tools

              list_prompts async 

              list_prompts() -> list[StaticPrompt]

              Get all available prompts from MCP servers.

              Source code in src/llmling_agent/mcp_server/manager.py 
              155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
              async def list_prompts(self) -> list[StaticPrompt]:
    """Get all available prompts from MCP servers."""
    prompts = []
    for client in self.clients.values():
        try:
            result = await client.list_prompts()
        except Exception:
            logger.exception("Failed to get prompts from MCP server")
        for prompt in result.prompts:
            try:
                converted = await convert_mcp_prompt(client, prompt)
                prompts.append(converted)
            except Exception:
                logger.exception("Failed to convert prompt: %s", prompt.name)
    return prompts

              list_resources async 

              list_resources() -> list[ResourceInfo]

              Get all available resources from MCP servers.

              Source code in src/llmling_agent/mcp_server/manager.py 
              171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
              async def list_resources(self) -> list[ResourceInfo]:
    """Get all available resources from MCP servers."""
    resources = []
    for client in self.clients.values():
        try:
            result = await client.list_resources()
        except Exception:
            logger.exception("Failed to get resources from MCP server")
        for resource in result.resources:
            try:
                converted = await convert_mcp_resource(resource)
                resources.append(converted)
            except Exception:
                logger.exception("Failed to convert resource: %s", resource.name)
    return resources

              setup_server async 

              setup_server(config: MCPServerConfig)

              Set up a single MCP server connection.

              Source code in src/llmling_agent/mcp_server/manager.py 
              112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
              async def setup_server(self, config: MCPServerConfig):
    """Set up a single MCP server connection."""
    if not config.enabled:
        return
    env = config.get_env_vars()
    match config:
        case StdioMCPServerConfig():
            client = MCPClient(transport_mode="stdio")
            client = await self.exit_stack.enter_async_context(client)
            await client.connect(config.command, args=config.args, env=env)
            client_id = f"{config.command}_{' '.join(config.args)}"
        case SSEMCPServerConfig():
            client = MCPClient(transport_mode="sse")
            client = await self.exit_stack.enter_async_context(client)
            await client.connect("", [], url=config.url, env=env)
            client_id = f"sse_{config.url}"
        case StreamableHTTPMCPServerConfig():
            client = MCPClient(transport_mode="streamable-http")
            client = await self.exit_stack.enter_async_context(client)
            await client.connect("", [], url=config.url, env=env)
            client_id = f"streamable_http_{config.url}"
    self.clients[client_id] = client

              convert_mcp_prompt async 

              convert_mcp_prompt(client: MCPClient, prompt: Prompt) -> StaticPrompt

              Convert MCP prompt to StaticPrompt.

              Source code in src/llmling_agent/mcp_server/manager.py 
              35
36
37
38
39
40
41
42
43
44
45
46
              async def convert_mcp_prompt(client: MCPClient, prompt: MCPPrompt) -> StaticPrompt:
    """Convert MCP prompt to StaticPrompt."""
    from mcp.types import EmbeddedResource, ImageContent

    result = await client.get_prompt(prompt.name)
    messages = [
        PromptMessage(role="system", content=message.content.text)
        for message in result.messages
        if not isinstance(message.content, EmbeddedResource | ImageContent)
    ]
    desc = prompt.description or "No description provided"
    return StaticPrompt(name=prompt.name, description=desc, messages=messages)

              convert_mcp_resource async 

              convert_mcp_resource(resource: Resource) -> ResourceInfo

              Convert MCP resource to ResourceInfo.

              Source code in src/llmling_agent/mcp_server/manager.py 
              49
50
51
52
53
              async def convert_mcp_resource(resource: MCPResource) -> ResourceInfo:
    """Convert MCP resource to ResourceInfo."""
    return ResourceInfo(
        name=resource.name, uri=str(resource.uri), description=resource.description
    )