my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Mnemosyne MCP Server — stdio and SSE transports.
#3
#4	Usage:
#5	# stdio (default) — for Claude Desktop, etc.
#6	mnemosyne mcp
#7
#8	# SSE — for web clients
#9	mnemosyne mcp --transport sse --port 8080
#10
#11	# Specific bank
#12	mnemosyne mcp --bank project_a
#13	"""
#14
#15	import os
#16	import sys
#17	import json
#18	import asyncio
#19	from typing import Optional
#20	from pathlib import Path
#21
#22	# Guarded import — MCP is optional
#23	try:
#24	from mcp.server import Server
#25	from mcp.server.stdio import stdio_server
#26	from mcp.types import TextContent, CallToolResult
#27	_MCP_AVAILABLE = True
#28	except ImportError:
#29	_MCP_AVAILABLE = False
#30	Server = None
#31	stdio_server = None
#32	TextContent = None
#33	CallToolResult = None
#34
#35	from mnemosyne.mcp_tools import get_tool_definitions, handle_tool_call
#36
#37	# ---------------------------------------------------------------------------
#38	# Server Setup
#39	# ---------------------------------------------------------------------------
#40
#41	async def _run_stdio() -> None:
#42	"""Run MCP server over stdio transport."""
#43	if not _MCP_AVAILABLE:
#44	raise RuntimeError("MCP not installed. Run: pip install mnemosyne-memory[mcp]")
#45
#46	server = Server("mnemosyne")
#47
#48	@server.list_tools()
#49	async def list_tools():
#50	return get_tool_definitions()
#51
#52	@server.call_tool()
#53	async def call_tool(name: str, arguments: dict) -> list:
#54	try:
#55	result = handle_tool_call(name, arguments)
#56	return [TextContent(type="text", text=json.dumps(result, indent=2, default=str))]
#57	except Exception as e:
#58	return [TextContent(type="text", text=json.dumps({"status": "error", "message": str(e)}, indent=2))]
#59
#60	async with stdio_server(server) as (read_stream, write_stream):
#61	await server.run(read_stream, write_stream, server.create_initialization_options())
#62
#63
#64	async def _run_sse(port: int = 8080) -> None:
#65	"""Run MCP server over SSE transport."""
#66	if not _MCP_AVAILABLE:
#67	raise RuntimeError("MCP not installed. Run: pip install mnemosyne-memory[mcp]")
#68
#69	# SSE transport requires additional imports
#70	try:
#71	from mcp.server.sse import SseServerTransport
#72	from starlette.applications import Starlette
#73	from starlette.routing import Route
#74	import uvicorn
#75	except ImportError:
#76	raise RuntimeError("SSE transport requires starlette and uvicorn. Run: pip install starlette uvicorn")
#77
#78	transport = SseServerTransport("/messages")
#79	server = Server("mnemosyne")
#80
#81	@server.list_tools()
#82	async def list_tools():
#83	return get_tool_definitions()
#84
#85	@server.call_tool()
#86	async def call_tool(name: str, arguments: dict) -> list:
#87	try:
#88	result = handle_tool_call(name, arguments)
#89	return [TextContent(type="text", text=json.dumps(result, indent=2, default=str))]
#90	except Exception as e:
#91	return [TextContent(type="text", text=json.dumps({"status": "error", "message": str(e)}, indent=2))]
#92
#93	async def handle_sse(request):
#94	async with transport.connect_sse(request.scope, request.receive, request.send) as streams:
#95	await server.run(streams[0], streams[1], server.create_initialization_options())
#96
#97	async def handle_messages(request):
#98	await transport.handle_post_message(request.scope, request.receive, request.send)
#99
#100	starlette_app = Starlette(
#101	routes=[
#102	Route("/sse", endpoint=handle_sse),
#103	Route("/messages", endpoint=handle_messages, methods=["POST"]),
#104	]
#105	)
#106
#107	config = uvicorn.Config(starlette_app, host="0.0.0.0", port=port, log_level="info")
#108	await uvicorn.Server(config).serve()
#109
#110
#111	# ---------------------------------------------------------------------------
#112	# CLI Entry Point
#113	# ---------------------------------------------------------------------------
#114
#115	def run_mcp_server(transport: str = "stdio", port: int = 8080, bank: Optional[str] = None) -> None:
#116	"""
#117	Run the Mnemosyne MCP server.
#118
#119	Args:
#120	transport: "stdio" or "sse"
#121	port: Port for SSE transport (ignored for stdio)
#122	bank: Default bank for operations (optional)
#123	"""
#124	if bank:
#125	os.environ["MNEMOSYNE_MCP_BANK"] = bank
#126
#127	if transport == "stdio":
#128	asyncio.run(_run_stdio())
#129	elif transport == "sse":
#130	asyncio.run(_run_sse(port))
#131	else:
#132	raise ValueError(f"Unknown transport: {transport}. Use 'stdio' or 'sse'.")
#133
#134
#135	def main(argv: Optional[list[str]] = None) -> None:
#136	"""CLI entry point for `mnemosyne mcp`."""
#137	import argparse
#138
#139	parser = argparse.ArgumentParser(description="Mnemosyne MCP Server")
#140	parser.add_argument(
#141	"--transport",
#142	choices=["stdio", "sse"],
#143	default="stdio",
#144	help="Transport protocol (default: stdio)"
#145	)
#146	parser.add_argument(
#147	"--port",
#148	type=int,
#149	default=8080,
#150	help="Port for SSE transport (default: 8080)"
#151	)
#152	parser.add_argument(
#153	"--bank",
#154	type=str,
#155	default=None,
#156	help="Default memory bank"
#157	)
#158	args = parser.parse_args(argv)
#159
#160	run_mcp_server(transport=args.transport, port=args.port, bank=args.bank)
#161
#162
#163	if __name__ == "__main__":
#164	main()
#165

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public