Lenses MCP Server

Implementation Reference

• src/lenses_mcp/tools/topics.py:268-305 (handler)

  The async handler function for the 'update_topic_metadata' tool, decorated with @mcp.tool() to register it. It takes environment and metadata dict, and posts to the API endpoint /api/v1/environments/{environment}/proxy/api/v1/metadata/topics to update the topic metadata.

  @mcp.tool()
  async def update_topic_metadata(
      environment: str, 
      metadata: Dict[str, Any]
  ) -> str:
      f"""
      Update topic metadata. The required parameters are: topicName, keyType and valueType.
      When updating tags, it is not a list of strings. 
      It is a list of objects with parameter 'name', e.g. [{{'name':'tag1'}},{{'name':'tag2'}}]
      
      Args:
          environment: The environment name.
          topic_name: Name of the topic.
          configs: Metadata key-value pairs with the following value types:
              {
                  "topicName": "text",
                  "keyType": "text",
                  "valueType": "text",
                  "keySchema": "text",
                  "keySchemaVersion": 1,
                  "keySchemaInlined": "text",
                  "valueSchema": "text",
                  "valueSchemaVersion": 1,
                  "valueSchemaInlined": "text",
                  "description": "text",
                  "tags": [
                      {{
                          "name": "text"
                      }}
                  ],
                  "additionalInfo": null
              }
      Returns:
          Success message.
      """
      endpoint = f"/api/v1/environments/{environment}/proxy/api/v1/metadata/topics"
      return await api_client._make_request("POST", endpoint, metadata)

• src/lenses_mcp/server.py:33-33 (registration)

  The call to register_topics(mcp) which executes the registration of the update_topic_metadata tool (along with other topic tools) via the nested @mcp.tool() decorators.

  register_topics(mcp)