X(Twitter) MCP Server

Instructions

Implementation Reference

• src/x_mcp/server.py:201-253 (handler)

  Executes the publish_draft tool: validates input, loads draft from file, publishes single tweet or thread to Twitter/X using Tweepy, removes draft, returns published tweet ID.

  async def handle_publish_draft(arguments: Any) -> Sequence[TextContent]:
      if not isinstance(arguments, dict) or "draft_id" not in arguments:
          raise ValueError("Invalid arguments for publish_draft")
      draft_id = arguments["draft_id"]
      filepath = os.path.join("drafts", draft_id)
      if not os.path.exists(filepath):
          raise ValueError(f"Draft {draft_id} does not exist")
      try:
          with open(filepath, "r") as f:
              draft = json.load(f)
          if "content" in draft:
              # Single tweet
              content = draft["content"]
              response = client.create_tweet(text=content)
              tweet_id = response.data['id']
              logger.info(f"Published tweet ID {tweet_id}")
              # Delete the draft after publishing
              os.remove(filepath)
              return [
                  TextContent(
                      type="text",
                      text=f"Draft {draft_id} published as tweet ID {tweet_id}",
                  )
              ]
          elif "contents" in draft:
              # Thread
              contents = draft["contents"]
              # Publish the thread
              last_tweet_id = None
              for content in contents:
                  if last_tweet_id is None:
                      response = client.create_tweet(text=content)
                  else:
                      response = client.create_tweet(text=content, in_reply_to_tweet_id=last_tweet_id)
                  last_tweet_id = response.data['id']
                  await asyncio.sleep(1)  # Avoid hitting rate limits
              logger.info(f"Published thread starting with tweet ID {last_tweet_id}")
              # Delete the draft after publishing
              os.remove(filepath)
              return [
                  TextContent(
                      type="text",
                      text=f"Draft {draft_id} published as thread starting with tweet ID {last_tweet_id}",
                  )
              ]
          else:
              raise ValueError(f"Invalid draft format for {draft_id}")
      except tweepy.TweepError as e:
          logger.error(f"Twitter API error: {e}")
          raise RuntimeError(f"Error publishing draft {draft_id}: {e}")
      except Exception as e:
          logger.error(f"Error publishing draft {draft_id}: {str(e)}")
          raise RuntimeError(f"Error publishing draft {draft_id}: {str(e)}")

• src/x_mcp/server.py:87-100 (registration)

  Registers the publish_draft tool in the @server.list_tools() handler, including name, description, and input schema.

  Tool(
      name="publish_draft",
      description="Publish a draft tweet or thread",
      inputSchema={
          "type": "object",
          "properties": {
              "draft_id": {
                  "type": "string",
                  "description": "ID of the draft to publish",
              },
          },
          "required": ["draft_id"],
      },
  ),

• src/x_mcp/server.py:90-99 (schema)

  Input schema definition for publish_draft tool: object with required 'draft_id' string property.

  inputSchema={
      "type": "object",
      "properties": {
          "draft_id": {
              "type": "string",
              "description": "ID of the draft to publish",
          },
      },
      "required": ["draft_id"],
  },

• src/x_mcp/server.py:126-127 (helper)

  Tool dispatch in @server.call_tool(): routes 'publish_draft' calls to the handle_publish_draft function.

  elif name == "publish_draft":
      return await handle_publish_draft(arguments)