orchestrata_ultimate.session.mcp_tools
Reference › orchestrata_ultimate › session › mcp_tools
SessionMCPTools exposes session management as MCP tools. It is built from a RedisSessionStore and a FactEmitter. tool_definitions returns the schema of each tool, and handle_tool_call(tool_name, arguments) routes a call to a private handler: session_create, session_get, session_close, session_emit_fact, session_query_facts and session_fact_count. An unknown tool name returns a dictionary with an error key, and failures are reported the same way instead of raising.
At a glance
| Source | src/orchestrata_ultimate/session/mcp_tools.py |
| Lines of code | 256 |
| Classes | 1 |
| Functions | 0 |
| Imports (internal) | orchestrata_ultimate.core.base.traces, orchestrata_ultimate.session.fact_emitter, orchestrata_ultimate.session.redis_store, orchestrata_ultimate.session.types |
| Imported by | none |
| Tests that import it | tests/generated_recovery/test_session_mcp_tools.py, tests/session/test_session_mcp_tools.py, tests/session/test_session_signatures.py |
Class diagram
Dependency diagram
API
Classes
SessionMCPTools
Decorators: traces(spec='006', story='US-003', reqs=['REQ-006-FR13-001']) · Line 27
MCP tool provider for session management operations.
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__(redis_store: RedisSessionStore, emitter: FactEmitter) -> None |
|
tool_definitions property |
tool_definitions() -> list[dict[str, Any]] |
Return MCP tool schema definitions. |
handle_tool_call |
handle_tool_call(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any] |
Route an MCP tool call to the appropriate handler. |
_create |
_create(args: dict[str, Any]) -> dict[str, Any] |
Create a new session. |
_get |
_get(args: dict[str, Any]) -> dict[str, Any] |
Retrieve session state. |
_close |
_close(args: dict[str, Any]) -> dict[str, Any] |
Close and delete a session. |
_emit_fact |
_emit_fact(args: dict[str, Any]) -> dict[str, Any] |
Emit a typed fact to a session. |
_query_facts |
_query_facts(args: dict[str, Any]) -> dict[str, Any] |
Query facts from a session. |
_fact_count |
_fact_count(args: dict[str, Any]) -> dict[str, Any] |
Get fact count for a session. |
Source
"""MCP tool definitions for session management.
Provides tool schemas and handler functions that expose session management
operations in MCP-compatible format. These can be registered with any
MCP server implementation.
No dependency on FastMCP or any external MCP library — provides a clean
interface that any MCP server can wrap.
"""
import logging
from typing import Any
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.fact_emitter import FactEmitter
from orchestrata_ultimate.session.redis_store import RedisSessionStore
from orchestrata_ultimate.session.types import FactType
logger = logging.getLogger(__name__)
@traces(
spec="006",
story="US-003",
reqs=["REQ-006-FR13-001"],
)
class SessionMCPTools:
"""MCP tool provider for session management operations.
Exposes session CRUD, fact emission, and query operations
as MCP-compatible tools with standardized schemas.
"""
def __init__(self, redis_store: RedisSessionStore, emitter: FactEmitter) -> None:
self._store = redis_store
self._emitter = emitter
@property
@traces(spec="006", story="FR-013", reqs=["REQ-006-FR13-004"])
def tool_definitions(self) -> list[dict[str, Any]]:
"""Return MCP tool schema definitions.
Each tool follows the MCP tool schema format with name,
description, and JSON Schema inputSchema.
"""
return [
{
"name": "session_create",
"description": "Create a new session with optional goals",
"inputSchema": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Unique session identifier",
},
"goals": {
"type": "array",
"items": {"type": "string"},
"description": "Session goals",
},
},
"required": ["session_id"],
},
},
{
"name": "session_get",
"description": "Get session state by ID",
"inputSchema": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session identifier to retrieve",
},
},
"required": ["session_id"],
},
},
{
"name": "session_close",
"description": "Close and delete a session",
"inputSchema": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session identifier to close",
},
},
"required": ["session_id"],
},
},
{
"name": "session_emit_fact",
"description": "Emit a typed fact to a session",
"inputSchema": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Target session identifier",
},
"fact_type": {
"type": "string",
"enum": [t.value for t in FactType],
"description": "Type of fact to emit",
},
"data": {
"type": "object",
"description": "Fact payload data",
},
"source": {
"type": "string",
"description": "Source of the fact",
},
},
"required": ["session_id", "fact_type", "data", "source"],
},
},
{
"name": "session_query_facts",
"description": "Query facts from a session, optionally filtered by type",
"inputSchema": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session to query facts from",
},
"fact_type": {
"type": "string",
"enum": [t.value for t in FactType],
"description": "Optional fact type filter",
},
},
"required": ["session_id"],
},
},
{
"name": "session_fact_count",
"description": "Get the number of facts in a session",
"inputSchema": {
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session to count facts for",
},
},
"required": ["session_id"],
},
},
]
@traces(spec="006", story="FR-010", reqs=["REQ-006-FR10-005", "REQ-006-FR13-005"])
def handle_tool_call(
self, tool_name: str, arguments: dict[str, Any]
) -> dict[str, Any]:
"""Route an MCP tool call to the appropriate handler.
Args:
tool_name: Name of the tool to invoke.
arguments: Tool arguments as a dictionary.
Returns:
Result dictionary. Contains an "error" key on failure.
"""
handlers = {
"session_create": self._create,
"session_get": self._get,
"session_close": self._close,
"session_emit_fact": self._emit_fact,
"session_query_facts": self._query_facts,
"session_fact_count": self._fact_count,
}
handler = handlers.get(tool_name)
if handler is None:
return {"error": f"Unknown tool: {tool_name}"}
return handler(arguments)
@traces(spec="006", story="FR-010", reqs=["REQ-006-FR10-001", "REQ-006-FR13-002"])
def _create(self, args: dict[str, Any]) -> dict[str, Any]:
"""Create a new session."""
session_id = args["session_id"]
goals = args.get("goals", [])
self._store.save_session(
{
"id": session_id,
"status": "active",
"goals": goals,
}
)
self._store.set_active(session_id)
return {"session_id": session_id, "status": "created"}
@traces(spec="006", story="FR-010", reqs=["REQ-006-FR10-003", "REQ-006-FR13-002"])
def _get(self, args: dict[str, Any]) -> dict[str, Any]:
"""Retrieve session state."""
session = self._store.get_session(args["session_id"])
if session is None:
return {"error": "Session not found", "session_id": args["session_id"]}
return session
@traces(spec="006", story="FR-010", reqs=["REQ-006-FR10-002", "REQ-006-FR13-002"])
def _close(self, args: dict[str, Any]) -> dict[str, Any]:
"""Close and delete a session."""
session_id = args["session_id"]
self._store.clear_active(session_id)
self._store.delete_session(session_id)
return {"session_id": session_id, "status": "closed"}
@traces(spec="006", story="FR-008", reqs=["REQ-006-FR8-001", "REQ-006-FR8-004"])
def _emit_fact(self, args: dict[str, Any]) -> dict[str, Any]:
"""Emit a typed fact to a session."""
session_id = args["session_id"]
fact_type = FactType(args["fact_type"])
fact = self._emitter.emit_typed(
session_id, fact_type, args["data"], args["source"]
)
return {
"session_id": session_id,
"fact_id": fact.id,
"status": "fact_emitted",
}
@traces(
spec="006",
story="US-003",
reqs=["REQ-006-FR13-003", "REQ-006-FR9-001", "REQ-006-US-003-005"],
)
def _query_facts(self, args: dict[str, Any]) -> dict[str, Any]:
"""Query facts from a session."""
session_id = args["session_id"]
fact_type = args.get("fact_type")
facts = self._emitter.query_facts(session_id, fact_type=fact_type)
return {
"session_id": session_id,
"count": len(facts),
"facts": [
{
"id": f.id,
"type": f.fact_type.value,
"data": f.data,
"source": f.source,
}
for f in facts
],
}
@traces(spec="006", story="FR-013", reqs=["REQ-006-FR13-003"])
def _fact_count(self, args: dict[str, Any]) -> dict[str, Any]:
"""Get fact count for a session."""
session_id = args["session_id"]
count = self._emitter.get_fact_count(session_id)
return {"session_id": session_id, "count": count}