MCP Observer Server

Instructions

Implementation Reference

• src/mcp_observer_server/server.py:175-185 (handler)

  Executes the subscribe tool: resolves path, adds current session to watched paths subscribers, notifies if new path, returns confirmation.

  if name == "subscribe":
      assert path_str is not None, "path_str should not be None for subscribe tool"
      p = Path(path_str).expanduser().resolve()
      is_new_path = p not in watched
      watched.setdefault(p, set()).add(session)
  
      # Send notification if this is a new path being watched
      if is_new_path:
          asyncio.create_task(send_resources_list_changed_notification())
  
      return [TextContent(type="text", text=f"Subscribed to {p}")]

• src/mcp_observer_server/server.py:62-70 (schema)

  Defines the input schema for the 'subscribe' tool, requiring a 'path' string parameter.

  Tool(
      name="subscribe",
      description="Subscribe to changes on a file or directory",
      inputSchema={
          "type": "object",
          "properties": {"path": {"type": "string"}},
          "required": ["path"],
      },
  ),

• src/mcp_observer_server/server.py:59-99 (registration)

  Registers the 'subscribe' tool (among others) by returning it in the list_tools() handler.

  @server.list_tools()
  async def list_tools() -> list[Tool]:
      return [
          Tool(
              name="subscribe",
              description="Subscribe to changes on a file or directory",
              inputSchema={
                  "type": "object",
                  "properties": {"path": {"type": "string"}},
                  "required": ["path"],
              },
          ),
          Tool(
              name="unsubscribe",
              description="Unsubscribe from changes on a file or directory",
              inputSchema={
                  "type": "object",
                  "properties": {"path": {"type": "string"}},
                  "required": ["path"],
              },
          ),
          Tool(
              name="list_watched",
              description="List all currently monitored paths and their subscriber counts",
              inputSchema={
                  "type": "object",
                  "properties": {},
                  "additionalProperties": False,
              },
          ),
          Tool(
              name="subscribe_default",
              description="Subscribe to the default watched.txt file for development",
              inputSchema={
                  "type": "object",
                  "properties": {},
                  "additionalProperties": False,
              },
          ),
      ]