postgres-mcp 0.3.2

A PostgreSQL MCP (Model Context Protocol) server implementation for building AI agents
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

import asyncio
import os
import subprocess
from typing import Optional

from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServer, MCPServerStdio


class PostgresMCPServer(MCPServerStdio):
    """PostgreSQL MCP Server implementation"""

    def __init__(self):
        postgres_mcp_path = "postgres-mcp"

        super().__init__(
            name="PostgreSQL MCP Server",
            params={
                "command": postgres_mcp_path,
                "args": ["stdio"],
            },
        )


async def run(mcp_server: MCPServer):
    agent = Agent(
        name="PostgreSQL Assistant",
        instructions="Use the tools to interact with PostgreSQL database and perform operations like querying, creating tables, and managing indexes.",
        mcp_servers=[mcp_server],
    )

    context = []
    context.append(
        "SYSTEM: The following questions are about the database server 'postgres://postgres:postgres@localhost:5432'. Please answer the questions in details with all the information the tool provides."
    )

    while True:
        # get user input from CLI
        question = input("Enter a question: ")
        context.append(f"USER: {question}")

        # run the agent
        try:
            result = await Runner.run(starting_agent=agent, input="\n\n".join(context))
            context.append(f"ASSISTANT: {result.final_output}")
        except Exception as e:
            continue

        print(result.final_output)


async def main():
    async with PostgresMCPServer() as server:
        trace_id = gen_trace_id()
        with trace(workflow_name="PostgreSQL MCP Agent", trace_id=trace_id):
            print(f"View trace: https://platform.openai.com/traces/{trace_id}\n")
            await run(server)


if __name__ == "__main__":
    asyncio.run(main())