Server#
ragbits.agents.mcp.server.MCPServer
#
Bases: ABC
Base class for Model Context Protocol servers.
connect
abstractmethod
async
#
Connect to the server. For example, this might mean spawning a subprocess or
opening a network connection. The server is expected to remain connected until cleanup()
is called.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
cleanup
abstractmethod
async
#
Cleanup the server. For example, this might mean closing a subprocess or closing a network connection.
list_tools
abstractmethod
async
#
call_tool
abstractmethod
async
#
ragbits.agents.mcp.server.MCPServerStdioParams
#
Bases: TypedDict
Mirrors mcp.client.stdio.StdioServerParameters
, but lets you pass params without another import.
command
instance-attribute
#
The executable to run to start the server. For example, python
or node
.
args
instance-attribute
#
Command line args to pass to the command
executable. For example, ['foo.py']
or
['server.js', '--port', '8080']
.
env
instance-attribute
#
The environment variables to set for the server. .
cwd
instance-attribute
#
The working directory to use when spawning the process.
encoding
instance-attribute
#
The text encoding used when sending/receiving messages to the server. Defaults to utf-8
.
encoding_error_handler
instance-attribute
#
The text encoding error handler. Defaults to strict
.
See https://docs.python.org/3/library/codecs.html#codec-base-classes for explanations of possible values.
ragbits.agents.mcp.server.MCPServerStdio
#
MCPServerStdio(params: MCPServerStdioParams, cache_tools_list: bool = False, name: str | None = None, client_session_timeout_seconds: float | None = 5)
Bases: _MCPServerWithClientSession
MCP server implementation that uses the stdio transport. See the spec for details.
Create a new MCP server based on the stdio transport.
PARAMETER | DESCRIPTION |
---|---|
params |
The params that configure the server. This includes the command to run to start the server, the args to pass to the command, the environment variables to set for the server, the working directory to use when spawning the process, and the text encoding used when sending/receiving messages to the server.
TYPE:
|
cache_tools_list |
Whether to cache the tools list. If
TYPE:
|
name |
A readable name for the server. If not provided, we'll create one from the command.
TYPE:
|
client_session_timeout_seconds |
the read timeout passed to the MCP ClientSession.
TYPE:
|
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
server_initialize_result
instance-attribute
#
client_session_timeout_seconds
instance-attribute
#
params
instance-attribute
#
params = StdioServerParameters(command=params['command'], args=get('args', []), env=get('env'), cwd=get('cwd'), encoding=get('encoding', 'utf-8'), encoding_error_handler=get('encoding_error_handler', 'strict'))
connect
async
#
Connect to the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
cleanup
async
#
Cleanup the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
list_tools
async
#
List the tools available on the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
call_tool
async
#
Invoke a tool on the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
invalidate_tools_cache
#
create_streams
#
create_streams() -> AbstractAsyncContextManager[tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage], GetSessionIdCallback | None]]
Create the streams for the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
ragbits.agents.mcp.server.MCPServerSseParams
#
Bases: TypedDict
Mirrors the params inmcp.client.sse.sse_client
.
ragbits.agents.mcp.server.MCPServerSse
#
MCPServerSse(params: MCPServerSseParams, cache_tools_list: bool = False, name: str | None = None, client_session_timeout_seconds: float | None = 5)
Bases: _MCPServerWithClientSession
MCP server implementation that uses the HTTP with SSE transport. See the spec for details.
Create a new MCP server based on the HTTP with SSE transport.
PARAMETER | DESCRIPTION |
---|---|
params |
The params that configure the server. This includes the URL of the server, the headers to send to the server, the timeout for the HTTP request, and the timeout for the SSE connection.
TYPE:
|
cache_tools_list |
Whether to cache the tools list. If
TYPE:
|
name |
A readable name for the server. If not provided, we'll create one from the URL.
TYPE:
|
client_session_timeout_seconds |
the read timeout passed to the MCP ClientSession.
TYPE:
|
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
server_initialize_result
instance-attribute
#
client_session_timeout_seconds
instance-attribute
#
connect
async
#
Connect to the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
cleanup
async
#
Cleanup the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
list_tools
async
#
List the tools available on the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
call_tool
async
#
Invoke a tool on the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
invalidate_tools_cache
#
create_streams
#
create_streams() -> AbstractAsyncContextManager[tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage], GetSessionIdCallback | None]]
Create the streams for the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
ragbits.agents.mcp.server.MCPServerStreamableHttpParams
#
Bases: TypedDict
Mirrors the params inmcp.client.streamable_http.streamablehttp_client
.
ragbits.agents.mcp.server.MCPServerStreamableHttp
#
MCPServerStreamableHttp(params: MCPServerStreamableHttpParams, cache_tools_list: bool = False, name: str | None = None, client_session_timeout_seconds: float | None = 5)
Bases: _MCPServerWithClientSession
MCP server implementation that uses the Streamable HTTP transport. See the spec for details.
Create a new MCP server based on the Streamable HTTP transport.
PARAMETER | DESCRIPTION |
---|---|
params |
The params that configure the server. This includes the URL of the server, the headers to send to the server, the timeout for the HTTP request, and the timeout for the Streamable HTTP connection and whether we need to terminate on close. |
cache_tools_list |
Whether to cache the tools list. If
TYPE:
|
name |
A readable name for the server. If not provided, we'll create one from the URL.
TYPE:
|
client_session_timeout_seconds |
the read timeout passed to the MCP ClientSession.
TYPE:
|
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
server_initialize_result
instance-attribute
#
client_session_timeout_seconds
instance-attribute
#
connect
async
#
Connect to the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
cleanup
async
#
Cleanup the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
list_tools
async
#
List the tools available on the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
call_tool
async
#
Invoke a tool on the server.
Source code in packages/ragbits-agents/src/ragbits/agents/mcp/server.py
invalidate_tools_cache
#
create_streams
#
create_streams() -> AbstractAsyncContextManager[tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage], GetSessionIdCallback | None]]
Create the streams for the server.