compliance_lookup

Instructions

Implementation Reference

• src/bay_street_mcp/server.py:61-80 (handler)

  The call_tool function handles the 'compliance_lookup' tool invocation. It validates the tool name, extracts query and top_k arguments, calls _store.search(), and returns results as JSON text content.

  @server.call_tool()
  async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
      if name != "compliance_lookup":
          raise ValueError(f"Unknown tool: {name}")
      if _store is None:
          raise RuntimeError("Compliance store not initialized")
  
      query = arguments["query"]
      top_k = min(int(arguments.get("top_k", 5)), 10)
  
      results = _store.search(query, top_k=top_k)
      if not results:
          return [
              TextContent(
                  type="text",
                  text="No relevant passages found. The store may be empty; run `bay-street-ingest` to load a regulation.",
              )
          ]
  
      return [TextContent(type="text", text=json.dumps(results, indent=2))]

• src/bay_street_mcp/server.py:27-58 (registration)

  The list_tools function registers the 'compliance_lookup' tool with its name, description, and input schema (query string + optional top_k integer).

  @server.list_tools()
  async def list_tools() -> list[Tool]:
      return [
          Tool(
              name="compliance_lookup",
              description=(
                  "Search Canadian financial-services regulations (OSFI, PIPEDA, "
                  "FINTRAC, Quebec Law 25) and return the most relevant passages "
                  "with citation metadata. Use this tool whenever the user asks "
                  "about Canadian financial regulation, compliance, AI/data "
                  "governance for Canadian financial institutions, AML/ATF, "
                  "operational risk, model risk, or related topics."
              ),
              inputSchema={
                  "type": "object",
                  "properties": {
                      "query": {
                          "type": "string",
                          "description": "Natural-language question about Canadian financial-services compliance.",
                      },
                      "top_k": {
                          "type": "integer",
                          "description": "Number of passages to return. Default 5, max 10.",
                          "default": 5,
                          "minimum": 1,
                          "maximum": 10,
                      },
                  },
                  "required": ["query"],
              },
          )
      ]

• src/bay_street_mcp/server.py:30-57 (schema)

  The Tool definition includes the input schema for compliance_lookup: required 'query' (string) and optional 'top_k' (integer, default 5, min 1, max 10).

  Tool(
      name="compliance_lookup",
      description=(
          "Search Canadian financial-services regulations (OSFI, PIPEDA, "
          "FINTRAC, Quebec Law 25) and return the most relevant passages "
          "with citation metadata. Use this tool whenever the user asks "
          "about Canadian financial regulation, compliance, AI/data "
          "governance for Canadian financial institutions, AML/ATF, "
          "operational risk, model risk, or related topics."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "Natural-language question about Canadian financial-services compliance.",
              },
              "top_k": {
                  "type": "integer",
                  "description": "Number of passages to return. Default 5, max 10.",
                  "default": 5,
                  "minimum": 1,
                  "maximum": 10,
              },
          },
          "required": ["query"],
      },
  )

• src/bay_street_mcp/store.py:47-63 (helper)

  The ComplianceStore.search() method performs the actual vector similarity search using ChromaDB, returning passages with citation metadata and distance scores.

  def search(self, query: str, top_k: int = 5) -> list[dict[str, Any]]:
      result = self.collection.query(
          query_texts=[query],
          n_results=top_k,
          include=["documents", "metadatas", "distances"],
      )
      documents = result["documents"][0] if result["documents"] else []
      metadatas = result["metadatas"][0] if result["metadatas"] else []
      distances = result["distances"][0] if result["distances"] else []
      return [
          {
              "passage": doc,
              "citation": meta,
              "distance": dist,
          }
          for doc, meta, dist in zip(documents, metadatas, distances, strict=False)
      ]

• src/bay_street_mcp/store.py:20-45 (helper)

  The ComplianceStore class manages a ChromaDB persistent client/collection, with load_default(), count(), and add() methods supporting the compliance_lookup tool.

  class ComplianceStore:
      def __init__(self, db_path: Path = DEFAULT_DB_PATH) -> None:
          db_path.mkdir(parents=True, exist_ok=True)
          self.client = chromadb.PersistentClient(
              path=str(db_path),
              settings=Settings(anonymized_telemetry=False),
          )
          self.collection = self.client.get_or_create_collection(
              name=COLLECTION_NAME,
              metadata={"hnsw:space": "cosine"},
          )
  
      @classmethod
      def load_default(cls) -> ComplianceStore:
          return cls()
  
      def count(self) -> int:
          return self.collection.count()
  
      def add(
          self,
          documents: list[str],
          ids: list[str],
          metadatas: list[dict[str, Any]],
      ) -> None:
          self.collection.add(documents=documents, ids=ids, metadatas=metadatas)