get_policy

Instructions

Implementation Reference

• openterms_mcp_server.py:291-316 (handler)

  The handle_tool function dispatches 'get_policy' by making a GET request to /v1/policy and formatting the response. It parses the active policy JSON, checks if active, then displays rules (max_amount_per_receipt, daily_spend_cap, etc.) in a human-readable format.

  elif name == "get_policy":
      resp = client.get("/v1/policy", headers=_headers())
      if resp.status_code == 200:
          policy = resp.json()
          if not policy.get("active") and policy.get("active") is not True:
              if policy.get("version", 0) == 0:
                  return "No active policy. All actions are allowed."
          rules = policy.get("rules", [])
          lines = [
              f"Active policy (version {policy.get('version', '?')}):",
              f"  Rules ({len(rules)}):"
          ]
          for i, rule in enumerate(rules):
              rtype = rule.get("type", "unknown")
              if rtype in ("max_amount_per_receipt", "daily_spend_cap", "max_action_context_keys"):
                  lines.append(f"    {i+1}. {rtype}: limit={rule.get('limit')}")
              elif rtype == "escalate_above_amount":
                  lines.append(f"    {i+1}. {rtype}: threshold={rule.get('threshold')}")
              elif rtype in ("allowed_action_types", "blocked_action_types"):
                  lines.append(f"    {i+1}. {rtype}: {rule.get('values')}")
              elif rtype == "required_terms_url_prefix":
                  lines.append(f"    {i+1}. {rtype}: {rule.get('prefix')}")
              else:
                  lines.append(f"    {i+1}. {rtype}: {json.dumps(rule)}")
          return "\n".join(lines)
      return _format_error(resp)

• openterms_mcp_server.py:79-88 (schema)

  Registration entry in the TOOLS list defining the 'get_policy' tool with description and an empty inputSchema (no parameters required).

  # --- MVP 2: Policy Tools ---
  {
      "name": "get_policy",
      "description": (
          "Get the active policy (guardrails) for this workspace. "
          "Returns the rules that govern what this agent is allowed to do. "
          "An agent SHOULD call this on startup to understand its constraints."
      ),
      "inputSchema": {"type": "object", "properties": {}},
  },

• openterms_mcp_server.py:445-452 (registration)

  MCP server registration: the tool is listed via list_tools() and dispatched via call_tool() which calls handle_tool(name, arguments).

  @server.list_tools()
  async def list_tools():
      return [types.Tool(**t) for t in TOOLS]
  
  @server.call_tool()
  async def call_tool(name: str, arguments: dict):
      result = handle_tool(name, arguments or {})
      return [types.TextContent(type="text", text=result)]

• app.py:602-609 (handler)

  API endpoint handler for GET /v1/policy. Called by the MCP server handler; retrieves the active policy from the database.

  @app.route('/v1/policy', methods=['GET'])
  @require_auth
  def get_policy():
      """Get the active policy profile for the workspace."""
      policy = db.get_active_policy(g.workspace_id)
      if not policy:
          return jsonify({'active': False, 'version': 0, 'rules': []})
      return jsonify(policy)

• db.py:500-513 (helper)

  Database helper function get_active_policy that queries the policy_profile table for the active policy of a workspace.

  def get_active_policy(workspace_id):
      """Get the active policy profile for a workspace, or None."""
      conn = get_db()
      row = conn.execute(
          "SELECT * FROM policy_profile WHERE workspace_id = ? AND active = 1",
          (workspace_id,)
      ).fetchone()
      conn.close()
      if row:
          result = dict(row)
          if isinstance(result.get('rules'), str):
              result['rules'] = json.loads(result['rules'])
          return result
      return None