MCP Observer Server

Instructions

Implementation Reference

• src/mcp_observer_server/server.py:71-79 (registration)

  Registration of the 'unsubscribe' tool, defining its name, description, and input schema requiring a 'path' parameter.

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

• src/mcp_observer_server/server.py:186-195 (handler)

  The handler logic within the call_tool function that executes the unsubscribe tool: removes the session from the watched path's subscribers and cleans up empty entries.

  elif name == "unsubscribe":
      assert path_str is not None, "path_str should not be None for unsubscribe tool"
      p = Path(path_str).expanduser().resolve()
      subs = watched.get(p)
      if subs and session in subs:
          subs.remove(session)
          if not subs:
              del watched[p]
          return [TextContent(type="text", text=f"Unsubscribed from {p}")]
      return [TextContent(type="text", text=f"Not subscribed to {p}")]

• src/mcp_observer_server/server.py:238-249 (handler)

  Additional resource-specific unsubscribe handler using the MCP resource unsubscribe protocol, with identical logic to the tool.

  @server.unsubscribe_resource()
  async def unsubscribe_resource_handler(uri: AnyUrl) -> None:
      if not uri.path:
          return
      p = Path(uri.path).resolve()
      session = server.request_context.session
      subs = watched.get(p)
      if subs and session in subs:
          subs.remove(session)
          if not subs:
              del watched[p]

• src/mcp_observer_server/server.py:74-78 (schema)

  Input schema for the unsubscribe tool, specifying an object with a required 'path' string property.

  inputSchema={
      "type": "object",
      "properties": {"path": {"type": "string"}},
      "required": ["path"],
  },