get_breaking

Instructions

Implementation Reference

• src/zipp_mcp/server.py:118-137 (handler)

  MCP tool handler for 'get_breaking'. Registered via @mcp.tool decorator with name='get_breaking' and a description of breaking news (importance >= 75, last 24h). The async function accepts 'lang' (default en-US) and 'limit' (default 10), then delegates to client.get_breaking().

  @mcp.tool(
      name="get_breaking",
      description=(
          "Breaking news only — last 24 hours, importance score >= 75. "
          "Lower volume than get_latest but every item is market-moving "
          "by Zipp's editorial threshold."
      ),
  )
  async def get_breaking(
      lang: str = _DEFAULT_LANG,
      limit: int = _DEFAULT_LIMIT,
  ) -> dict[str, Any]:
      """Get breaking news (importance >= 75) from the last 24h.
  
      Args:
          lang: BCP-47 language tag (default en-US).
          limit: Max results (1-30, default 10).
      """
      async with ZippClient() as client:
          return await client.get_breaking(lang=lang, limit=limit)

• src/zipp_mcp/client.py:107-113 (helper)

  ZippClient.get_breaking() — the underlying HTTP helper that calls GET /api/v1/news/breaking with lang and limit query parameters, returning the raw JSON dict.

  async def get_breaking(
      self,
      *,
      lang: str = "en-US",
      limit: int = 10,
  ) -> dict[str, Any]:
      return await self._get("/breaking", params={"lang": lang, "limit": limit})

• src/zipp_mcp/server.py:42-55 (registration)

  Registration infrastructure: _build_mcp() creates the FastMCP instance, assigned to module-level 'mcp'. The @mcp.tool decorator on line 118 registers 'get_breaking' as an MCP tool.

  def _build_mcp() -> FastMCP:
      settings = get_settings()
      return FastMCP(
          name="Zipp",
          instructions=_INSTRUCTIONS,
          host=settings.mcp_host,
          port=settings.mcp_port,
          transport_security=TransportSecuritySettings(
              enable_dns_rebinding_protection=False,
          ),
      )
  
  
  mcp = _build_mcp()

• tests/test_client.py:77-90 (helper)

  Test verifying the get_breaking client method constructs the correct path /api/v1/news/breaking with query params lang=tr-TR and limit=3.

  async def test_get_breaking_path(settings: Settings) -> None:
      seen: list[httpx.Request] = []
  
      def handler(req: httpx.Request) -> httpx.Response:
          seen.append(req)
          return httpx.Response(200, json={"items": []})
  
      async with ZippClient(settings=settings, transport=httpx.MockTransport(handler)) as c:
          await c.get_breaking(lang="tr-TR", limit=3)
  
      assert seen[0].url.path == "/api/v1/news/breaking"
      params = parse_qs(seen[0].url.query.decode())
      assert params["lang"] == ["tr-TR"]
      assert params["limit"] == ["3"]

• tests/test_server.py:28-38 (registration)

  Test verifying that 'get_breaking' is among the registered tool names on the MCP server.

  @pytest.mark.parametrize(
      "tool_name",
      [
          "search",
          "get_latest",
          "get_breaking",
          "get_featured",
          "get_post",
          "list_categories",
      ],
  )