OpenAI SDK example

Author: mortensi

README.md

@@ -64,6 +64,28 @@ To configure this Redis MCP Server, consider the following environment variables
 | `REDIS_CERT_REQS`       | Whether the client should verify the server's certificate | `"required"`  |
 | `REDIS_CA_CERTS`        | Path to the trusted CA certificates file                  | None          |
 
+## Integration with OpenAI Agents SDK
+
+Integrate this MCP Server with the OpenAI Agents SDK. Read the [documents](https://openai.github.io/openai-agents-python/mcp/) to learn more about the integration of the SDK with MCP.
+
+Install the Python SDK.
+
+```commandline
+pip install openai-agents
+```
+
+Configure the OpenAI token:
+
+```commandline
+export OPENAI_API_KEY="<openai_token>"
+```
+
+And run the [application](./examples/redis_assistant.py).
+
+```commandline
+python3.13 redis_assistant.py 
+```
+
 ## Integration with Claude Desktop
 You can configure Claude Desktop to use this MCP Server.
 

examples/redis_assistant.py

@@ -0,0 +1,73 @@
+import asyncio
+from agents import Agent, Runner
+from openai.types.responses import ResponseTextDeltaEvent
+from agents.mcp import MCPServerStdio
+from collections import deque
+
+
+# Set up and create the agent
+async def build_agent():
+    # Redis MCP Server
+    server = MCPServerStdio(
+        params={
+            "command": "uv",
+            "args": [
+                "--directory", "<path_to_mcp_server>/mcp-redis/src/",
+                "run", "main.py"
+            ],
+        }
+    )
+
+    await server.connect()
+
+    # Create and return the agent
+    agent = Agent(
+        name="Redis Assistant",
+        instructions="You are a helpful assistant capable of reading and writing to Redis.",
+        mcp_servers=[server]
+    )
+
+    return agent
+
+
+# CLI interaction
+async def cli(agent, max_history=30):
+    print("🔧 Redis Assistant CLI — Ask me something (type 'exit' to quit):\n")
+    conversation_history = deque(maxlen=max_history)
+
+    while True:
+        q = input("❓> ")
+        if q.strip().lower() in {"exit", "quit"}:
+            break
+
+        if (len(q.strip()) > 0):
+            # Add the user's message to history
+            conversation_history.append({"role": "user", "content": q})
+
+            # Format the context into a single string
+            context = ""
+            for turn in conversation_history:
+                prefix = "User" if turn["role"] == "user" else "Assistant"
+                context += f"{prefix}: {turn['content']}\n"
+
+            result = Runner.run_streamed(agent, context.strip())
+
+            response_text = ""
+            async for event in result.stream_events():
+                if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
+                    print(event.data.delta, end="", flush=True)
+                    response_text += event.data.delta
+            print("\n")
+
+            # Store assistant's reply in history
+            conversation_history.append({"role": "assistant", "content": response_text})
+
+
+# Main entry point
+async def main():
+    agent = await build_agent()
+    await cli(agent)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())