homer_search_calls

Instructions

Implementation Reference

• src/opensips_mcp/tools/ecosystem_tools.py:48-82 (handler)

  The MCP tool handler function for 'homer_search_calls'. Decorated with @mcp.tool() and @require_permission('mi.read'). Accepts caller, callee, call_id, time_from, time_to as optional parameters, then delegates to HomerClient.search_calls().

  @mcp.tool()
  @require_permission("mi.read")
  async def homer_search_calls(
      ctx: Context,
      caller: str | None = None,
      callee: str | None = None,
      call_id: str | None = None,
      time_from: str | None = None,
      time_to: str | None = None,
  ) -> dict[str, Any]:
      """Search SIP calls captured by Homer/SIPCAPTURE.
  
      Parameters
      ----------
      caller:
          Filter by calling party (From user).
      callee:
          Filter by called party (To user).
      call_id:
          Filter by SIP Call-ID.
      time_from:
          Start of time range (ISO-8601 string).
      time_to:
          End of time range (ISO-8601 string).
      """
      app = ctx.request_context.lifespan_context
      return await HomerClient.search_calls(
          app.http_client,
          _homer_url(app),
          caller=caller,
          callee=callee,
          call_id=call_id,
          time_from=time_from,
          time_to=time_to,
      )

• src/opensips_mcp/ecosystem/homer.py:22-84 (helper)

  The HomerClient.search_calls() static method that builds the JSON body and POSTs to Homer's /api/v3/call/transaction endpoint to search captured SIP calls.

  async def search_calls(
      http_client: httpx.AsyncClient,
      homer_url: str,
      *,
      caller: str | None = None,
      callee: str | None = None,
      call_id: str | None = None,
      time_from: str | None = None,
      time_to: str | None = None,
  ) -> dict[str, Any]:
      """Search captured SIP calls in Homer.
  
      Parameters
      ----------
      http_client:
          Shared httpx async client.
      homer_url:
          Homer base URL (e.g. ``http://homer:9080``).
      caller:
          Filter by calling party (From user).
      callee:
          Filter by called party (To user).
      call_id:
          Filter by SIP Call-ID.
      time_from:
          Start of time range (ISO-8601 or epoch).
      time_to:
          End of time range (ISO-8601 or epoch).
      """
      base = homer_url.rstrip("/")
      search_body: dict[str, Any] = {
          "param": {
              "transaction": {"call": True},
              "limit": 200,
              "search": {},
          },
      }
  
      search_params = search_body["param"]["search"]
      if caller:
          search_params["from_user"] = caller
      if callee:
          search_params["to_user"] = callee
      if call_id:
          search_params["callid"] = call_id
  
      if time_from:
          search_body["timestamp"] = search_body.get("timestamp", {})
          search_body["timestamp"]["from"] = time_from
      if time_to:
          search_body["timestamp"] = search_body.get("timestamp", {})
          search_body["timestamp"]["to"] = time_to
  
      try:
          resp = await http_client.post(f"{base}{_SEARCH_PATH}", json=search_body)
          resp.raise_for_status()
          return resp.json()  # type: ignore[no-any-return]
      except httpx.HTTPStatusError as exc:
          logger.warning("Homer search failed: %s", exc)
          return {"error": str(exc), "status_code": exc.response.status_code}
      except httpx.HTTPError as exc:
          logger.warning("Homer search connection error: %s", exc)
          return {"error": str(exc)}

• src/opensips_mcp/resources/ecosystem_resources.py:12-18 (schema)

  Tool documentation resource entry for 'homer_search_calls' describing parameters, permission, and example usage.

  _TOOL_DOCS: dict[str, dict[str, str]] = {
      "homer_search_calls": {
          "description": "Search SIP calls captured by Homer/SIPCAPTURE",
          "parameters": "caller, callee, call_id, time_from, time_to",
          "permission": "mi.read",
          "example": 'homer_search_calls(caller="alice", time_from="2026-01-01T00:00:00Z")',
      },

• src/opensips_mcp/tools/ecosystem_tools.py:423-423 (registration)

  Registration of 'homer_search_calls' in the available_tools listing under the 'homer' category for the ecosystem_overview tool.

  "homer": ["homer_search_calls", "homer_get_call_flow", "homer_stats"],